[NRL-1351] Finish off CLI script to manage NRL pointer-type permissions

Author: mattdean3-nhs

scripts/manage_permissions.py

@@ -13,8 +13,15 @@
     "NRL_AUTH_BUCKET_NAME", f"nhsd-nrlf--{nrl_env}-authorization-store"
 )
 
+COMPARE_AND_CONFIRM = (
+    True
+    if nrl_env == "prod"
+    else os.getenv("COMPARE_AND_CONFIRM", "false").lower() == "true"
+)
+
 print(f"Using NRL environment: {nrl_env}")
 print(f"Using NRL auth bucket: {nrl_auth_bucket_name}")
+print(f"Compare and confirm mode: {COMPARE_AND_CONFIRM}")
 print()
 
 
@@ -33,7 +40,7 @@ def _list_s3_keys(file_key_prefix: str) -> list[str]:
     }
 
     page_iterator = paginator.paginate(**params)
-    keys = []
+    keys: list[str] = []
     for page in page_iterator:
         if "Contents" in page:
             keys.extend([item["Key"] for item in page["Contents"]])
@@ -61,19 +68,26 @@ def _get_perms_from_s3(file_key: str) -> str | None:
     return item["Body"].read().decode("utf-8")
 
 
-def list_apps() -> list[str]:
+def list_apps() -> None:
+    """
+    List all applications in the NRL environment.
+    """
     keys = _list_s3_keys("")
-    apps = [key.split("/")[0] for key in keys]
+    apps = set([key.split("/")[0] for key in keys])
 
     if not apps:
         print("No applications found in the bucket.")
         return []
 
-    print(f"Listing all {len(apps)} apps in bucket...")
-    return apps
+    print(f"There are {len(apps)} apps in {nrl_env} env:")
+    for app in apps:
+        print(f"- {app}")
 
 
-def list_orgs(app_id: str) -> list[str]:
+def list_orgs(app_id: str) -> None:
+    """
+    List all organizations for a specific application.
+    """
     keys = _list_s3_keys(f"{app_id}/")
     orgs = [
         key.split("/", maxsplit=2)[1].removesuffix(".json")
@@ -83,23 +97,36 @@ def list_orgs(app_id: str) -> list[str]:
 
     if not orgs:
         print(f"No organizations found for app {app_id}.")
-        return []
 
-    print(f"Listing {len(orgs)} organizations for {app_id}...")
-    return orgs
+    print(f"There are {len(orgs)} organizations for app {app_id}:")
+    for org in orgs:
+        print(f"- {org}")
+
+
+def list_allowed_types() -> None:
+    """
+    List all pointer types that can be used in permissions.
+    """
+    print("The following pointer-types are allowed:")
+
+    for pointer_type, attributes in TYPE_ATTRIBUTES.items():
+        print(f"- %-45s (%s)" % (pointer_type, attributes["display"][:45]))
 
 
-def get(app_id: str, org_ods: str) -> list[str]:
+def show_perms(app_id: str, org_ods: str) -> None:
+    """
+    Show the permissions for a specific application and organization.
+    """
     perms = _get_perms_from_s3(f"{app_id}/{org_ods}.json")
 
     if not perms:
         print(f"No permissions file found for {app_id}/{org_ods}.")
-        return []
+        return
 
     pointertype_perms = json.loads(perms)
     if not pointertype_perms:
         print(f"No pointer-types found in permission file for {app_id}/{org_ods}.")
-        return []
+        return
 
     type_data = {
         pointertype_perm: TYPE_ATTRIBUTES.get(
@@ -108,27 +135,63 @@ def get(app_id: str, org_ods: str) -> list[str]:
         for pointertype_perm in pointertype_perms
     }
     types = [
-        f"{type_data[pointertype_perm]['display']} ({pointertype_perm})"
+        "%-45s (%s)" % (type_data[pointertype_perm]["display"][:44], pointertype_perm)
         for pointertype_perm in pointertype_perms
     ]
 
-    print(f"The current permissions for {app_id}/{org_ods} are:")
-    return types
+    print(f"{app_id}/{org_ods} is allowed to access these pointer-types:")
+    for type_display in types:
+        print(f"- {type_display}")
 
 
-def set(app_id: str, org_ods: str, *pointer_types: str) -> list[str]:
+def set_perms(app_id: str, org_ods: str, *pointer_types: str) -> None:
+    """
+    Set permissions for an application and organization to access specific pointer types.
+    """
     if not pointer_types:
         print(
             "No pointer types provided. Please specify at least one pointer type or use clear_perms command."
         )
-        return []
+        return
+
+    if len(pointer_types) == 1 and pointer_types[0] == "all":
+        print("Setting permissions for access to all pointer types.")
+        pointer_types = tuple(TYPE_ATTRIBUTES.keys())
 
     unknown_types = [pt for pt in pointer_types if pt not in TYPE_ATTRIBUTES]
     if unknown_types:
-        print(f"Warning: Unknown pointer types provided: {', '.join(unknown_types)}")
+        print(f"Error: Unknown pointer types provided: {', '.join(unknown_types)}")
         print()
+        return
 
     permissions_content = json.dumps(pointer_types, indent=4)
+
+    if COMPARE_AND_CONFIRM:
+        current_perms = _get_perms_from_s3(f"{app_id}/{org_ods}.json")
+        if current_perms == permissions_content:
+            print(
+                f"No changes needed for {app_id}/{org_ods}. Current permissions match the new ones."
+            )
+            return
+
+        print()
+        print(f"Current permissions for {app_id}/{org_ods}:")
+        print(current_perms if current_perms else "No permissions set.")
+
+        print()
+        print("New permissions to be set to:")
+        print(f"{permissions_content}")
+
+        print()
+        confirm = (
+            input("Do you want to proceed with these changes? (yes/NO): ")
+            .strip()
+            .lower()
+        )
+        if confirm != "yes":
+            print("Operation cancelled at user request.")
+            return
+
     s3 = _get_s3_client()
     s3.put_object(
         Bucket=nrl_auth_bucket_name,
@@ -137,10 +200,18 @@ def set(app_id: str, org_ods: str, *pointer_types: str) -> list[str]:
         ContentType="application/json",
     )
 
-    return get(app_id, org_ods)
+    print()
+    print(f"Set permissions for {app_id}/{org_ods}")
 
+    print()
+    show_perms(app_id, org_ods)
 
-def clear(app_id: str, org_ods: str) -> None:
+
+def clear_perms(app_id: str, org_ods: str) -> None:
+    """
+    Clear permissions for an application and organization.
+    This will remove all permissions for the specified app and org.
+    """
     s3 = _get_s3_client()
     s3.put_object(
         Bucket=nrl_auth_bucket_name,
@@ -152,4 +223,13 @@ def clear(app_id: str, org_ods: str) -> None:
 
 
 if __name__ == "__main__":
-    fire.Fire()
+    fire.Fire(
+        {
+            "list_apps": list_apps,
+            "list_orgs": list_orgs,
+            "list_allowed_types": list_allowed_types,
+            "show_perms": show_perms,
+            "set_perms": set_perms,
+            "clear_perms": clear_perms,
+        }
+    )