from os import environ from hydra_client import HydraAdmin from graphqlclient import GraphQLClient import urllib import json GRAPHQL_URL = environ['GRAPHQL_URL'] GRAPHQL_CLIENT = GraphQLClient(GRAPHQL_URL) class User(): def __init__(self, username): self.username = username try: self._load_remote_user_info() except urllib.error.HTTPError as error: raise BackendConnectionError( error.code, error.headers, ("Error during retrieval of userdata - " + error.reason)) def _load_remote_user_info(self): querystring = '''{{ getUser(username: "{0}"){{ email, applications{{ edges{{ node{{ name }} }} }}, roles{{ edges{{ node{{ name }} }} }} }}}}'''.format(self.username) result = json.loads(GRAPHQL_CLIENT.execute(querystring)) data = result["data"]["getUser"] self.applications = list(map(lambda x: x["node"]["name"], data["applications"]["edges"])) self.roles = list(map(lambda x: x["node"]["name"], data["roles"]["edges"])) self.email = data["email"] def has_app_permission(self, appname): return appname in self.applications def get_oauth_session(self): """Create openID Connect token Use the userdata stored in the user object to create an OpenID Connect token. The token will be passed to Hydra, which will store it and serve it to all OpenID Connect Clients, that successfully query the /userinfo endpoint. Every field in the "id_token" dictionary can be accessed through standard scopes and claims. See https://openid.net/specs/openid-connect-core-1_0.html#Claims Returns: OpenID Connect token of type dict """ return { "id_token": { "name": self.username, "preferred_username": self.username, "email" : self.email, "openappstack_roles": self.roles} } class BackendConnectionError(Exception): """Raised when requests to the backend server fail Attributes: code -- http response code headers -- http response headers reason -- reson for the error """ def __init__(self, code, headers, reason): self.code = code self.headers = headers self.reason = reason