Merge pull request #7 from ministryofjustice/rbac-uplift

Rbac uplift

.gitignore

@@ -159,4 +159,11 @@ cython_debug/
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 .idea/
 # VSCode Config
-.vscode
+.vscode
+
+.secrets
+.vars
+/rbac
+
+*.ldif
+*.ldif.j2

cli/__init__.py

@@ -1,28 +1,33 @@
 import click
-from cli import ldap
+import cli.ldap.add_roles_to_username, cli.ldap.rbac
 
 from cli import git
+import cli.env
+
 
 @click.group()
 def main_group():
     pass
 
+
 @click.command()
 @click.option("--user-ou", help="OU to add users to, defaults to ou=Users", default="ou=Users")
 @click.option("--root-dn", help="Root DN to add users to", default="dc=moj,dc=com")
 @click.argument("user-role-list", required=True)
 def add_roles_to_users(user_ou, root_dn, user_role_list):
-    ldap.process_user_roles_list(user_role_list, user_ou, root_dn)
+    cli.ldap.add_roles_to_username.process_user_roles_list(user_role_list, user_ou, root_dn)
 
 
 @click.command()
-def git_test():
-    git.dl_test()
+@click.option("--rbac-repo-tag", help="RBAC repo tag to use", default="master")
+def rbac_uplift(rbac_repo_tag):
+    cli.ldap.rbac.main(rbac_repo_tag)
+
 
 # from cli.ldap import test
 
 main_group.add_command(add_roles_to_users)
-main_group.add_command(git_test)
+main_group.add_command(rbac_uplift)
 
 if __name__ == "__main__":
     main_group()

cli/env.py

@@ -0,0 +1,48 @@
+import os
+
+from dotenv import dotenv_values
+
+import ast
+
+# ldap_host = os.getenv("LDAP_HOST")
+# ldap_user = os.getenv("LDAP_USER")
+# ldap_password = os.getenv("LDAP_PASSWORD")
+# db_user = os.getenv("DB_USER")
+# db_password = os.getenv("DB_PASSWORD")
+# db_host = os.getenv("DB_HOST")
+# db_port = os.getenv("DB_PORT")
+# db_service_name = os.getenv("DB_SERVICE_NAME")
+# gh_app_id = os.getenv("GH_APP_ID")
+# gh_private_key = os.getenv("GH_PRIVATE_KEY")
+# gh_installation_id = os.getenv("GH_INSTALLATION_ID")
+
+
+vars = {
+    **{
+        key.replace("VAR_", "").replace("_DICT", ""): ast.literal_eval(val) if "DICT" in key else val
+        for key, val in dotenv_values(".vars").items()
+        if val is not None
+    },  # load development variables
+    **{
+        key.replace("VAR_", "").replace("_DICT", ""): ast.literal_eval(val) if "DICT" in key else val
+        for key, val in os.environ.items()
+        if key.startswith("VAR_") and val is not None
+    },
+}
+# loads all environment variables starting with SECRET_ into a dictionary
+secrets = {
+    **{
+        key.replace("SECRET_", "").replace("_DICT", "").replace("SSM_", ""): ast.literal_eval(val)
+        if "_DICT" in key
+        else val
+        for key, val in dotenv_values(".secrets").items()
+        if val is not None
+    },
+    **{
+        key.replace("SECRET_", "").replace("_DICT", "").replace("SSM_", ""): ast.literal_eval(val)
+        if "DICT" in key
+        else val
+        for key, val in os.environ.items()
+        if key.startswith("SECRET_") or key.startswith("SSM_") and val is not None
+    },
+}

cli/git/__init__.py

@@ -4,47 +4,37 @@
 import time
 import requests
 import logging
-from cli import config
+from cli import env
+
+
 def get_access_token(app_id, private_key, installation_id):
     # Create a JSON Web Token (JWT) using the app's private key
     now = int(time.time())
-    payload = {
-        "iat": now,
-        "exp": now + 600,
-        "iss": app_id
-    }
+    payload = {"iat": now, "exp": now + 600, "iss": app_id}
     jwt_token = jwt.encode(payload, private_key, algorithm="RS256")
 
     # Exchange the JWT for an installation access token
-    headers = {
-        "Authorization": f"Bearer {jwt_token}",
-        "Accept": "application/vnd.github.v3+json"
-    }
-    response = requests.post(f"https://api.github.com/app/installations/{installation_id}/access_tokens",
-                             headers=headers)
+    headers = {"Authorization": f"Bearer {jwt_token}", "Accept": "application/vnd.github.v3+json"}
+    response = requests.post(
+        f"https://api.github.com/app/installations/{installation_id}/access_tokens", headers=headers
+    )
     # extract the token from the response
     access_token = response.json().get("token")
     return access_token
 
-def get_repo(url, token=None, auth_type="x-access-token", dest_name="repo"):
+
+def get_repo(url, depth="1", branch_or_tag="master", token=None, auth_type="x-access-token", dest_name="repo"):
     # if there is an @ in the url, assume auth is already specified
-    if '@' in url:
-        logging.info('auth already specified in url')
-        return Repo.clone_from(url, dest_name)
+    multi_options = ["--depth " + depth, "--branch " + branch_or_tag]
+    if "@" in url:
+        logging.info("auth already specified in url")
+        return Repo.clone_from(url, dest_name, multi_options=multi_options)
     # if there is a token, assume auth is required and use the token and auth_type
     elif token:
         templated_url = f'https://{auth_type}:{token}@{url.split("//")[1]}'
-        logging.info(f'cloning with token: {templated_url}')
-        return Repo.clone_from(templated_url, dest_name)
+        logging.info(f"cloning with token: {templated_url}")
+        return Repo.clone_from(templated_url, dest_name, multi_options=multi_options)
     # if there is no token, assume auth is not required and clone without
     else:
-        logging.info('cloning without auth')
-        return Repo.clone_from(url, dest_name)
-def dl_test():
-    app_id = config.gh_app_id
-    private_key = config.gh_private_key
-    installation_id = config.gh_installation_id
-    url = 'https://github.com/ministryofjustice/hmpps-delius-pipelines.git'
-    token = get_access_token(app_id, private_key, installation_id)
-    repo = get_repo(url, token=token, dest_name='delius-pipelines')
-    print(repo)
+        logging.info("cloning without auth")
+        return Repo.clone_from(url, dest_name, multi_options=multi_options)

cli/ldap/__init__.py

@@ -1,14 +1,12 @@
 from ldap3 import Server, Connection, ALL
-
-# import oracledb
-import logging
-
-logging.basicConfig(level=logging.DEBUG)
+from logging import log
 
 
+# import oracledb
 def ldap_connect(ldap_host, ldap_user, ldap_password):
+    server = Server(ldap_host, get_info=ALL)
     return Connection(
-        server=ldap_host, user=ldap_user, password=ldap_password, auto_bind="NO_TLS", authentication="SIMPLE"
+        server=server, user=ldap_user, password=ldap_password, auto_bind="NO_TLS", authentication="SIMPLE"
     )
 
 

cli/ldap/add_roles_to_username.py

@@ -1,16 +1,22 @@
-import logging
-
-from cli import config
+from cli.logging import log
+from cli import env
 from cli.ldap import ldap_connect
 
 
 def parse_user_role_list(user_role_list):
+    # The format of the list should be a pipe separated list of username and role lists,
+    # where the username and role list is separated by a comma character,
+    # and the roles are separated by a semi-colon:
+    # username1,role1;role2;role3|username2,role1;role2
+
     return {user.split(",")[0]: user.split(",")[1].split(";") for user in user_role_list.split("|")}
 
 
 def add_roles_to_user(username, roles, user_ou="ou=Users", root_dn="dc=moj,dc=com"):
-    logging.info(f"Adding roles {roles} to user {username}")
-    ldap_connection = ldap_connect(config.ldap_host, config.ldap_user, config.ldap_password)
+    log.info(f"Adding roles {roles} to user {username}")
+    ldap_connection = ldap_connect(
+        env.vars.get("LDAP_HOST"), env.vars.get("LDAP_USER"), env.secrets.get("LDAP_BIND_PASSWORD")
+    )
     for role in roles:
         ldap_connection.add(
             f"cn={role},cn={username},{user_ou},{root_dn}",
@@ -30,6 +36,7 @@ def add_roles_to_user(username, roles, user_ou="ou=Users", root_dn="dc=moj,dc=co
 
 
 def process_user_roles_list(user_role_list, user_ou="ou=Users", root_dn="dc=moj,dc=com"):
+    log.info(f"secrets: {env.secrets}")
     user_roles = parse_user_role_list(user_role_list)
     for user, roles in user_roles.items():
         add_roles_to_user(user, roles, user_ou, root_dn)