re-formatted with Black

Pipfile

@@ -22,6 +22,7 @@ pyyaml = "*"
 ldap3 = "*"
 apscheduler = "*"
 python-dotenv = "*"
+msal = "*"
 
 [pipenv]
 allow_prereleases = true

app.py

@@ -20,22 +20,17 @@ scheduler.start()
 atexit.register(lambda: scheduler.shutdown(wait=False))
 
 
-@github_app.on('team.created')
+@github_app.on("team.created")
 def sync_new_team():
     """
     Sync a new team when it is created
     :return:
     """
-    owner = github_app.payload['organization']['login']
-    team_id = github_app.payload['team']['id']
-    slug = github_app.payload['team']['slug']
+    owner = github_app.payload["organization"]["login"]
+    team_id = github_app.payload["team"]["id"]
+    slug = github_app.payload["team"]["slug"]
     client = github_app.installation_client
-    sync_team(
-        client=client,
-        owner=owner,
-        team_id=team_id,
-        slug=slug
-    )
+    sync_team(client=client, owner=owner, team_id=team_id, slug=slug)
 
 
 def sync_team(client=None, owner=None, team_id=None, slug=None):
@@ -53,32 +48,22 @@ def sync_team(client=None, owner=None, team_id=None, slug=None):
     directory_group = custom_map[slug] if slug in custom_map else slug
     directory_members = directory_group_members(group=directory_group)
     team_members = github_team_members(
-        client=client,
-        owner=owner,
-        team_id=team_id,
-        attribute='username'
+        client=client, owner=owner, team_id=team_id, attribute="username"
     )
     compare = compare_members(
-        group=directory_members,
-        team=team_members,
-        attribute='username'
+        group=directory_members, team=team_members, attribute="username"
     )
     if TEST_MODE:
-        print('Skipping execution due to TEST_MODE...')
+        print("Skipping execution due to TEST_MODE...")
         pprint(compare)
     else:
         try:
-            execute_sync(
-                org=org,
-                team=team,
-                slug=slug,
-                state=compare
-            )
+            execute_sync(org=org, team=team, slug=slug, state=compare)
         except ValueError as e:
-            if strtobool(os.environ['OPEN_ISSUE_ON_FAILURE']):
+            if strtobool(os.environ["OPEN_ISSUE_ON_FAILURE"]):
                 open_issue(client=client, slug=slug, message=e)
         except AssertionError as e:
-            if strtobool(os.environ['OPEN_ISSUE_ON_FAILURE']):
+            if strtobool(os.environ["OPEN_ISSUE_ON_FAILURE"]):
                 open_issue(client=client, slug=slug, message=e)
 
 
@@ -107,7 +92,7 @@ def github_team_info(client=None, owner=None, team_id=None):
     return org.team(team_id)
 
 
-def github_team_members(client=None, owner=None, team_id=None, attribute='username'):
+def github_team_members(client=None, owner=None, team_id=None, attribute="username"):
     """
     Look up members of a give team in GitHub
     :param client:
@@ -122,19 +107,22 @@ def github_team_members(client=None, owner=None, team_id=None, attribute='userna
     """
     team_members = []
     team = github_team_info(client=client, owner=owner, team_id=team_id)
-    if attribute == 'email':
+    if attribute == "email":
         for m in team.members():
             user = client.user(m.login)
-            team_members.append({'username': str(user.login).casefold(),
-                                 'email': str(user.email).casefold()})
+            team_members.append(
+                {
+                    "username": str(user.login).casefold(),
+                    "email": str(user.email).casefold(),
+                }
+            )
     else:
         for member in team.members():
-            team_members.append({'username': str(member).casefold(),
-                                 'email': ''})
+            team_members.append({"username": str(member).casefold(), "email": ""})
     return team_members
 
 
-def compare_members(group, team, attribute='username'):
+def compare_members(group, team, attribute="username"):
     """
     Compare users in GitHub and the User Directory to see which users need to be added or removed
     :param group:
@@ -148,12 +136,9 @@ def compare_members(group, team, attribute='username'):
     add_users = list(set(directory_list) - set(directory_list))
     remove_users = list(set(github_list) - set(directory_list))
     sync_state = {
-        'directory': group,
-        'github': team,
-        'action': {
-            'add': add_users,
-            'remove': remove_users
-        }
+        "directory": group,
+        "github": team,
+        "action": {"add": add_users, "remove": remove_users},
     }
     return sync_state
 
@@ -167,28 +152,28 @@ def execute_sync(org, team, slug, state):
     :param state:
     :return:
     """
-    total_changes = len(state['action']['remove']) + len(state['action']['add'])
-    if len(state['directory']) == 0:
+    total_changes = len(state["action"]["remove"]) + len(state["action"]["add"])
+    if len(state["directory"]) == 0:
         message = f"{os.environ.get('USER_DIRECTORY', 'LDAP').upper()} group returned empty: {slug}"
         raise ValueError(message)
-    elif int(total_changes) > int(os.environ.get('CHANGE_THRESHOLD', 25)):
+    elif int(total_changes) > int(os.environ.get("CHANGE_THRESHOLD", 25)):
         message = "Skipping sync for {}.<br>".format(slug)
         message += "Total number of changes ({}) would exceed the change threshold ({}).".format(
-            str(total_changes), str(os.environ.get('CHANGE_THRESHOLD', 25))
+            str(total_changes), str(os.environ.get("CHANGE_THRESHOLD", 25))
         )
         message += "<br>Please investigate this change and increase your threshold if this is accurate."
         raise AssertionError(message)
     else:
-        for user in state['action']['add']:
+        for user in state["action"]["add"]:
             # Validate that user is in org
             if org.is_member(user):
-                pprint(f'Adding {user} to {slug}')
+                pprint(f"Adding {user} to {slug}")
                 team.add_or_update_membership(user)
             else:
-                pprint(f'Skipping {user} as they are not part of the org')
+                pprint(f"Skipping {user} as they are not part of the org")
 
-        for user in state['action']['remove']:
-            pprint(f'Removing {user} from {slug}')
+        for user in state["action"]["remove"]:
+            pprint(f"Removing {user} from {slug}")
             team.revoke_membership(user)
 
 
@@ -200,20 +185,20 @@ def open_issue(client, slug, message):
     :param message: Error message to detail
     :return:
     """
-    repo_for_issues = os.environ['REPO_FOR_ISSUES']
-    owner = repo_for_issues.split('/')[0]
-    repository = repo_for_issues.split('/')[1]
-    assignee = os.environ['ISSUE_ASSIGNEE']
+    repo_for_issues = os.environ["REPO_FOR_ISSUES"]
+    owner = repo_for_issues.split("/")[0]
+    repository = repo_for_issues.split("/")[1]
+    assignee = os.environ["ISSUE_ASSIGNEE"]
     client.create_issue(
         owner=owner,
         repository=repository,
         assignee=assignee,
         title="Team sync failed for @{}/{}".format(owner, slug),
-        body=str(message)
+        body=str(message),
     )
 
 
-def load_custom_map(file='syncmap.yml'):
+def load_custom_map(file="syncmap.yml"):
     """
     Custom team synchronization
     :param file:
@@ -222,15 +207,18 @@ def load_custom_map(file='syncmap.yml'):
     syncmap = {}
     if os.path.isfile(file):
         from yaml import load, Loader
-        with open(file, 'r') as f:
+
+        with open(file, "r") as f:
             data = load(f, Loader=Loader)
-        for d in data['mapping']:
-            syncmap[d['github']] = d['directory']
+        for d in data["mapping"]:
+            syncmap[d["github"]] = d["directory"]
 
     return syncmap
 
 
-@scheduler.scheduled_job(trigger=CronTrigger.from_crontab(CRON_INTERVAL), id='sync_all_teams')
+@scheduler.scheduled_job(
+    trigger=CronTrigger.from_crontab(CRON_INTERVAL), id="sync_all_teams"
+)
 def sync_all_teams():
     """
     Lookup teams in a GitHub org and synchronize all teams with your user directory
@@ -243,19 +231,19 @@ def sync_all_teams():
         installations = gh.app_client.app_installations
         for i in installations():
             client = gh.app_installation(installation_id=i.id)
-            org = client.organization(i.account['login'])
+            org = client.organization(i.account["login"])
             for team in org.teams():
                 sync_team(
                     client=client,
-                    owner=i.account['login'],
+                    owner=i.account["login"],
                     team_id=team.id,
-                    slug=team.slug
+                    slug=team.slug,
                 )
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     sync_all_teams()
     app.run(
-        host=os.environ.get('FLASK_RUN_HOST', '0.0.0.0'),
-        port=os.environ.get('FLASK_RUN_PORT', '5000')
+        host=os.environ.get("FLASK_RUN_HOST", "0.0.0.0"),
+        port=os.environ.get("FLASK_RUN_PORT", "5000"),
     )

githubapp/__init__.py

@@ -9,7 +9,7 @@ elif os.environ.get("USER_DIRECTORY", "LDAP").upper() == "AAD":
     from .azuread import AzureAD as DirectoryClient
 from .version import __version__
 
-__all__ = ['GitHubApp', 'DirectoryClient']
+__all__ = ["GitHubApp", "DirectoryClient"]
 
 # Set default logging handler to avoid "No handler found" warnings.
 import logging
@@ -22,13 +22,13 @@ rootlogger = logging.getLogger(__name__)
 if rootlogger.level == logging.NOTSET:
     rootlogger.setLevel(logging.WARN)
 
-CRON_INTERVAL = os.environ.get('SYNC_SCHEDULE', '0 * * * *')
+CRON_INTERVAL = os.environ.get("SYNC_SCHEDULE", "0 * * * *")
 # CHANGE_THRESHOLD = os.environ.get('CHANGE_THRESHOLD', 25)
 # REPO_FOR_ISSUES = os.environ.get('REPO_FOR_ISSUES')
 # ISSUE_ASSIGNEE = os.environ.get('ISSUE_ASSIGNEE')
 # OPEN_ISSUE_ON_FAILURE = strtobool(os.environ.get('OPEN_ISSUE_ON_FAILURE', 'False'))
 try:
-    TEST_MODE = strtobool(os.environ.get('TEST_MODE', 'False'))
+    TEST_MODE = strtobool(os.environ.get("TEST_MODE", "False"))
 except ValueError as e:
     rootlogger.warn('TEST_MODE should be set to "true" or "false"')
     rootlogger.warn(e)

githubapp/azuread.py

@@ -41,15 +41,18 @@ import msal
 
 LOG = logging.getLogger(__name__)
 
+
 class AzureAD:
     def __init__(self):
-        self.AZURE_TENANT_ID = os.environ['AZURE_TENANT_ID']
-        self.AZURE_CLIENT_ID = os.environ['AZURE_CLIENT_ID']
-        self.AZURE_CLIENT_SECRET = os.environ['AZURE_CLIENT_SECRET']
-        self.AZURE_APP_SCOPE = [f'https://graph.microsoft.com/.{x}' for x in os.environ['AZURE_APP_SCOPE'].split(" ")]
-        self.AZURE_API_ENDPOINT = os.environ['AZURE_API_ENDPOINT']
-        self.USERNAME_ATTRIBUTE = os.environ['USERNAME_ATTRIBUTE']
-
+        self.AZURE_TENANT_ID = os.environ["AZURE_TENANT_ID"]
+        self.AZURE_CLIENT_ID = os.environ["AZURE_CLIENT_ID"]
+        self.AZURE_CLIENT_SECRET = os.environ["AZURE_CLIENT_SECRET"]
+        self.AZURE_APP_SCOPE = [
+            f"https://graph.microsoft.com/.{x}"
+            for x in os.environ["AZURE_APP_SCOPE"].split(" ")
+        ]
+        self.AZURE_API_ENDPOINT = os.environ["AZURE_API_ENDPOINT"]
+        self.USERNAME_ATTRIBUTE = os.environ["USERNAME_ATTRIBUTE"]
 
     def get_access_token(self):
         """
@@ -58,25 +61,29 @@ class AzureAD:
         """
         app = msal.ConfidentialClientApplication(
             self.AZURE_CLIENT_ID,
-            authority=f'https://login.microsoftonline.com/{self.AZURE_TENANT_ID}',
-            client_credential=self.AZURE_CLIENT_SECRET
+            authority=f"https://login.microsoftonline.com/{self.AZURE_TENANT_ID}",
+            client_credential=self.AZURE_CLIENT_SECRET,
         )
 
         # Lookup the token in cache
         result = app.acquire_token_silent(self.AZURE_APP_SCOPE, account=None)
 
         if not result:
-            logging.info("No suitable token exists in cache. Let's get a new one from AAD.")
+            logging.info(
+                "No suitable token exists in cache. Let's get a new one from AAD."
+            )
             result = app.acquire_token_for_client(scopes=self.AZURE_APP_SCOPE)
 
         if "access_token" in result:
             print("Successfully authenticated!")
-            return result['access_token']
+            return result["access_token"]
 
         else:
             print(result.get("error"))
             print(result.get("error_description"))
-            print(result.get("correlation_id"))  # You may need this when reporting a bug
+            print(
+                result.get("correlation_id")
+            )  # You may need this when reporting a bug
 
     def get_group_members(self, token=None, group=None):
         """
@@ -88,22 +95,21 @@ class AzureAD:
         member_list = []
         # Calling graph using the access token
         graph_data = requests.get(  # Use token to call downstream service
-            self.AZURE_API_ENDPOINT,
-            headers={'Authorization': f'Bearer {token}'}
+            self.AZURE_API_ENDPOINT, headers={"Authorization": f"Bearer {token}"}
         ).json()
         # print("Graph API call result: %s" % json.dumps(graph_data, indent=2))
         groups = json.loads(json.dumps(graph_data, indent=2))
-        for group in groups['value']:
+        for group in groups["value"]:
             members = requests.get(
                 f'{self.AZURE_API_ENDPOINT}/groups/{group["id"]}/members',
-                headers={'Authorization': f'Bearer {token}'}
+                headers={"Authorization": f"Bearer {token}"},
             ).json()
-            for member in members['value']:
+            for member in members["value"]:
                 member_list.append(member[self.USERNAME_ATTRIBUTE])
         return member_list
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     aad = AzureAD()
     token = aad.get_access_token()
-    aad.get_group_members(token=token, group="0aafe564-0044-424c-8a75-e8a59e6a83d5")
+    aad.get_group_members(token=token, group="github-demo")

githubapp/core.py

@@ -10,8 +10,8 @@ from github3 import GitHub, GitHubEnterprise
 
 LOG = logging.getLogger(__name__)
 
-STATUS_FUNC_CALLED = 'HIT'
-STATUS_NO_FUNC_CALLED = 'MISS'
+STATUS_FUNC_CALLED = "HIT"
+STATUS_NO_FUNC_CALLED = "MISS"
 
 
 class GitHubApp(object):
@@ -33,12 +33,12 @@ class GitHubApp(object):
 
     @staticmethod
     def load_env(app):
-        app.config['GITHUBAPP_ID'] = int(os.environ['APP_ID'])
-        app.config['GITHUBAPP_SECRET'] = os.environ['WEBHOOK_SECRET']
-        if 'GHE_HOST' in os.environ:
-            app.config['GITHUBAPP_URL'] = 'https://{}'.format(os.environ['GHE_HOST'])
-        with open(os.environ['PRIVATE_KEY_PATH'], 'rb') as key_file:
-            app.config['GITHUBAPP_KEY'] = key_file.read()
+        app.config["GITHUBAPP_ID"] = int(os.environ["APP_ID"])
+        app.config["GITHUBAPP_SECRET"] = os.environ["WEBHOOK_SECRET"]
+        if "GHE_HOST" in os.environ:
+            app.config["GITHUBAPP_URL"] = "https://{}".format(os.environ["GHE_HOST"])
+        with open(os.environ["PRIVATE_KEY_PATH"], "rb") as key_file:
+            app.config["GITHUBAPP_KEY"] = key_file.read()
 
     def init_app(self, app):
         """
@@ -72,63 +72,67 @@ class GitHubApp(object):
             Default: '/'
         """
         self.load_env(app)
-        required_settings = ['GITHUBAPP_ID', 'GITHUBAPP_KEY', 'GITHUBAPP_SECRET']
+        required_settings = ["GITHUBAPP_ID", "GITHUBAPP_KEY", "GITHUBAPP_SECRET"]
         for setting in required_settings:
             if not app.config.get(setting):
-                raise RuntimeError("Flask-GitHubApp requires the '%s' config var to be set" % setting)
+                raise RuntimeError(
+                    "Flask-GitHubApp requires the '%s' config var to be set" % setting
+                )
 
-        app.add_url_rule(app.config.get('GITHUBAPP_ROUTE', '/'),
-                         view_func=self._flask_view_func,
-                         methods=['POST'])
+        app.add_url_rule(
+            app.config.get("GITHUBAPP_ROUTE", "/"),
+            view_func=self._flask_view_func,
+            methods=["POST"],
+        )
 
     @property
     def id(self):
-        return current_app.config['GITHUBAPP_ID']
+        return current_app.config["GITHUBAPP_ID"]
 
     @property
     def key(self):
-        key = current_app.config['GITHUBAPP_KEY']
-        if hasattr(key, 'encode'):
-            key = key.encode('utf-8')
+        key = current_app.config["GITHUBAPP_KEY"]
+        if hasattr(key, "encode"):
+            key = key.encode("utf-8")
         return key
 
     @property
     def secret(self):
-        secret = current_app.config['GITHUBAPP_SECRET']
-        if hasattr(secret, 'encode'):
-            secret = secret.encode('utf-8')
+        secret = current_app.config["GITHUBAPP_SECRET"]
+        if hasattr(secret, "encode"):
+            secret = secret.encode("utf-8")
         return secret
 
     @property
     def _api_url(self):
-        return current_app.config['GITHUBAPP_URL']
+        return current_app.config["GITHUBAPP_URL"]
 
     @property
     def client(self):
         """Unauthenticated GitHub client"""
-        if current_app.config.get('GITHUBAPP_URL'):
-            return GitHubEnterprise(current_app.config['GITHUBAPP_URL'])
+        if current_app.config.get("GITHUBAPP_URL"):
+            return GitHubEnterprise(current_app.config["GITHUBAPP_URL"])
         return GitHub()
 
     @property
     def payload(self):
         """GitHub hook payload"""
-        if request and request.json and 'installation' in request.json:
+        if request and request.json and "installation" in request.json:
             return request.json
 
-        raise RuntimeError('Payload is only available in the context of a GitHub hook request')
+        raise RuntimeError(
+            "Payload is only available in the context of a GitHub hook request"
+        )
 
     @property
     def installation_client(self):
         """GitHub client authenticated as GitHub app installation"""
         ctx = _app_ctx_stack.top
         if ctx is not None:
-            if not hasattr(ctx, 'githubapp_installation'):
+            if not hasattr(ctx, "githubapp_installation"):
                 client = self.client
                 client.login_as_app_installation(
-                    self.key,
-                    self.id,
-                    self.payload['installation']['id']
+                    self.key, self.id, self.payload["installation"]["id"]
                 )
                 ctx.githubapp_installation = client
             return ctx.githubapp_installation
@@ -138,10 +142,9 @@ class GitHubApp(object):
         """GitHub client authenticated as GitHub app"""
         ctx = _app_ctx_stack.top
         if ctx is not None:
-            if not hasattr(ctx, 'githubapp_app'):
+            if not hasattr(ctx, "githubapp_app"):
                 client = self.client
-                client.login_as_app(self.key,
-                                    self.id)
+                client.login_as_app(self.key, self.id)
                 ctx.githubapp_app = client
             return ctx.githubapp_app
 
@@ -163,15 +166,11 @@ class GitHubApp(object):
         """GitHub client authenticated as GitHub app installation"""
         ctx = _app_ctx_stack.top
         if installation_id is None:
-            raise RuntimeError('Installation ID is not specified.')
+            raise RuntimeError("Installation ID is not specified.")
         if ctx is not None:
-            if not hasattr(ctx, 'githubapp_installation'):
+            if not hasattr(ctx, "githubapp_installation"):
                 client = self.client
-                client.login_as_app_installation(
-                    self.key,
-                    self.id,
-                    installation_id
-                )
+                client.login_as_app_installation(self.key, self.id, installation_id)
                 ctx.githubapp_installation = client
             return ctx.githubapp_installation
 
@@ -211,8 +210,8 @@ class GitHubApp(object):
         functions_to_call = []
         calls = {}
 
-        event = request.headers['X-GitHub-Event']
-        action = request.json.get('action')
+        event = request.headers["X-GitHub-Event"]
+        action = request.json.get("action")
 
         self._verify_webhook()
 
@@ -220,7 +219,7 @@ class GitHubApp(object):
             functions_to_call += self._hook_mappings[event]
 
         if action:
-            event_action = '.'.join([event, action])
+            event_action = ".".join([event, action])
             if event_action in self._hook_mappings:
                 functions_to_call += self._hook_mappings[event_action]
 
@@ -230,22 +229,21 @@ class GitHubApp(object):
             status = STATUS_FUNC_CALLED
         else:
             status = STATUS_NO_FUNC_CALLED
-        return jsonify({'status': status,
-                        'calls': calls})
+        return jsonify({"status": status, "calls": calls})
 
     def _verify_webhook(self):
         if True:
             return
 
-        hub_signature = 'X-HUB-SIGNATURE'
+        hub_signature = "X-HUB-SIGNATURE"
         if hub_signature not in request.headers:
-            LOG.warning('Github Hook Signature not found.')
+            LOG.warning("Github Hook Signature not found.")
             abort(400)
 
-        signature = request.headers[hub_signature].split('=')[1]
+        signature = request.headers[hub_signature].split("=")[1]
 
-        mac = hmac.new(self.secret, msg=request.data, digestmod='sha1')
+        mac = hmac.new(self.secret, msg=request.data, digestmod="sha1")
 
         if not hmac.compare_digest(mac.hexdigest(), signature):
-            LOG.warning('GitHub hook signature verification failed.')
-            abort(400)
+            LOG.warning("GitHub hook signature verification failed.")
+            abort(400)

githubapp/ldap.py

@@ -10,39 +10,38 @@ LOG = logging.getLogger(__name__)
 class LDAPClient:
     def __init__(self):
         # Read settings from the config file and store them as constants
-        self.LDAP_SERVER_HOST = os.environ['LDAP_SERVER_HOST']
-        self.LDAP_SERVER_PORT = os.environ['LDAP_SERVER_PORT']
-        self.LDAP_BASE_DN = os.environ['LDAP_BASE_DN']
-        self.LDAP_USER_BASE_DN = os.environ['LDAP_USER_BASE_DN']
-        self.LDAP_USER_ATTRIBUTE = os.environ['LDAP_USER_ATTRIBUTE']
-        self.LDAP_USER_FILTER = os.environ['LDAP_USER_FILTER'].replace(
-            '{ldap_user_attribute}',
-            self.LDAP_USER_ATTRIBUTE
+        self.LDAP_SERVER_HOST = os.environ["LDAP_SERVER_HOST"]
+        self.LDAP_SERVER_PORT = os.environ["LDAP_SERVER_PORT"]
+        self.LDAP_BASE_DN = os.environ["LDAP_BASE_DN"]
+        self.LDAP_USER_BASE_DN = os.environ["LDAP_USER_BASE_DN"]
+        self.LDAP_USER_ATTRIBUTE = os.environ["LDAP_USER_ATTRIBUTE"]
+        self.LDAP_USER_FILTER = os.environ["LDAP_USER_FILTER"].replace(
+            "{ldap_user_attribute}", self.LDAP_USER_ATTRIBUTE
         )
-        self.LDAP_USER_MAIL_ATTRIBUTE = os.environ['LDAP_USER_MAIL_ATTRIBUTE']
-        self.LDAP_GROUP_BASE_DN = os.environ['LDAP_GROUP_BASE_DN']
-        self.LDAP_GROUP_FILTER = os.environ['LDAP_GROUP_FILTER']
-        self.LDAP_GROUP_MEMBER_ATTRIBUTE = os.environ['LDAP_GROUP_MEMBER_ATTRIBUTE']
-        if 'LDAP_BIND_USER' in os.environ:
-            self.LDAP_BIND_USER = os.environ['LDAP_BIND_USER']
-        elif 'LDAP_BIND_DN' in os.environ:
-            self.LDAP_BIND_USER = os.environ['LDAP_BIND_DN']
+        self.LDAP_USER_MAIL_ATTRIBUTE = os.environ["LDAP_USER_MAIL_ATTRIBUTE"]
+        self.LDAP_GROUP_BASE_DN = os.environ["LDAP_GROUP_BASE_DN"]
+        self.LDAP_GROUP_FILTER = os.environ["LDAP_GROUP_FILTER"]
+        self.LDAP_GROUP_MEMBER_ATTRIBUTE = os.environ["LDAP_GROUP_MEMBER_ATTRIBUTE"]
+        if "LDAP_BIND_USER" in os.environ:
+            self.LDAP_BIND_USER = os.environ["LDAP_BIND_USER"]
+        elif "LDAP_BIND_DN" in os.environ:
+            self.LDAP_BIND_USER = os.environ["LDAP_BIND_DN"]
         else:
-            raise Exception('LDAP credentials have not been specified')
-        if 'LDAP_PAGE_SIZE' in os.environ:
-            self.LDAP_PAGE_SIZE = os.environ['LDAP_SEARCH_PAGE_SIZE']
+            raise Exception("LDAP credentials have not been specified")
+        if "LDAP_PAGE_SIZE" in os.environ:
+            self.LDAP_PAGE_SIZE = os.environ["LDAP_SEARCH_PAGE_SIZE"]
         else:
             self.LDAP_PAGE_SIZE = 1000
-        if 'LDAP_BIND_PASSWORD' in os.environ:
-            self.LDAP_BIND_PASSWORD = os.environ['LDAP_BIND_PASSWORD']
+        if "LDAP_BIND_PASSWORD" in os.environ:
+            self.LDAP_BIND_PASSWORD = os.environ["LDAP_BIND_PASSWORD"]
         else:
-            raise Exception('LDAP credentials have not been specified')
+            raise Exception("LDAP credentials have not been specified")
         self.conn = Connection(
             self.LDAP_SERVER_HOST,
             user=self.LDAP_BIND_USER,
             password=self.LDAP_BIND_PASSWORD,
             auto_bind=True,
-            auto_range=True
+            auto_range=True,
         )
 
     def get_group_members(self, group_name):
@@ -56,16 +55,13 @@ class LDAPClient:
         member_list = []
         entries = self.conn.extend.standard.paged_search(
             search_base=self.LDAP_BASE_DN,
-            search_filter=self.LDAP_GROUP_FILTER.replace(
-                '{group_name}',
-                group_name
-            ),
+            search_filter=self.LDAP_GROUP_FILTER.replace("{group_name}", group_name),
             attributes=[self.LDAP_GROUP_MEMBER_ATTRIBUTE],
-            paged_size=self.LDAP_PAGE_SIZE
+            paged_size=self.LDAP_PAGE_SIZE,
         )
         for entry in entries:
-            if entry['type'] == 'searchResEntry':
-                for member in entry['attributes'][self.LDAP_GROUP_MEMBER_ATTRIBUTE]:
+            if entry["type"] == "searchResEntry":
+                for member in entry["attributes"][self.LDAP_GROUP_MEMBER_ATTRIBUTE]:
                     if self.LDAP_GROUP_BASE_DN in member:
                         pass
                         # print("Nested groups are not yet supported.")
@@ -77,9 +73,15 @@ class LDAPClient:
                         try:
                             member_dn = self.get_user_info(member)
                             # pprint(member_dn)
-                            username = str(member_dn['attributes'][self.LDAP_USER_ATTRIBUTE][0]).casefold()
-                            email = str(member_dn['attributes'][self.LDAP_USER_MAIL_ATTRIBUTE][0]).casefold()
-                            user_info = {'username': username, 'email': email}
+                            username = str(
+                                member_dn["attributes"][self.LDAP_USER_ATTRIBUTE][0]
+                            ).casefold()
+                            email = str(
+                                member_dn["attributes"][self.LDAP_USER_MAIL_ATTRIBUTE][
+                                    0
+                                ]
+                            ).casefold()
+                            user_info = {"username": username, "email": email}
                             member_list.append(user_info)
                         except Exception as e:
                             print(e)
@@ -93,7 +95,7 @@ class LDAPClient:
         :return:
         :rtype:
         """
-        if any(attr in member.casefold() for attr in ['uid=', 'cn=']):
+        if any(attr in member.casefold() for attr in ["uid=", "cn="]):
             search_base = member
         else:
             search_base = self.LDAP_USER_BASE_DN
@@ -101,8 +103,8 @@ class LDAPClient:
             try:
                 self.conn.search(
                     search_base=search_base,
-                    search_filter=self.LDAP_USER_FILTER.replace('{username}', member),
-                    attributes=["*"]
+                    search_filter=self.LDAP_USER_FILTER.replace("{username}", member),
+                    attributes=["*"],
                 )
                 data = json.loads(self.conn.entries[0].entry_to_json())
                 return data

githubapp/version.py

@@ -1 +1 @@
-__version__ = '0.2.0'
+__version__ = "0.2.0"