Feature/create identity center user

controlpanel/api/aws.py

@@ -1425,3 +1425,157 @@ def get_table(self, database_name, table_name, catalog_id=None):
             raise error
 
         return response
+
+
+class AWSSSOAdmin(AWSService):
+
+    def __init__(self, assume_role_name=None, profile_name=None, region_name=None):
+        super().__init__(assume_role_name, profile_name, region_name)
+        region = region_name or settings.AWS_DEFAULT_REGION
+        self.client = self.boto3_session.client("sso-admin", region_name=region)
+        self.identity_store_id = None
+
+    def get_identity_store_id(self):
+
+        if self.identity_store_id:
+            return self.identity_store_id
+
+        response = self.client.list_instances()
+        self.identity_store_id = response["Instances"][0]["IdentityStoreId"]
+        return self.identity_store_id
+
+
+class AWSIdentityStore(AWSService):
+
+    def __init__(self, assume_role_name=None, profile_name=None, region_name=None):
+        super().__init__(assume_role_name, profile_name, region_name)
+        region = region_name or settings.AWS_DEFAULT_REGION
+        self.client = self.boto3_session.client("identitystore", region_name=region)
+        self.sso_client = AWSSSOAdmin(
+            assume_role_name=assume_role_name, profile_name=profile_name, region_name=region_name
+        )
+
+    def get_user_id(self, user_email):
+        try:
+            response = self.client.get_user_id(
+                IdentityStoreId=self.sso_client.get_identity_store_id(),
+                AlternateIdentifier={
+                    "UniqueAttribute": {"AttributePath": "userName", "AttributeValue": user_email}
+                },
+            )
+
+            return response["UserId"]
+        except self.client.exceptions.ResourceNotFoundException:
+            return None
+
+    def get_group_id(self, group_name):
+        try:
+            response = self.client.get_group_id(
+                IdentityStoreId=self.sso_client.get_identity_store_id(),
+                AlternateIdentifier={
+                    "UniqueAttribute": {
+                        "AttributePath": "displayName",
+                        "AttributeValue": group_name,
+                    }
+                },
+            )
+
+            return response["GroupId"]
+        except self.client.exceptions.ResourceNotFoundException as error:
+            log.exception(error.response["Error"]["Message"])
+            raise error
+
+    def get_group_membership_id(self, group_name, user_email):
+        try:
+            response = self.client.get_group_membership_id(
+                IdentityStoreId=self.sso_client.get_identity_store_id(),
+                GroupId=self.get_group_id(group_name),
+                MemberId={"UserId": self.get_user_id(user_email)},
+            )
+
+            return response["MembershipId"]
+        except self.client.exceptions.ResourceNotFoundException as error:
+            log.info(error.response["Error"]["Message"])
+            raise None
+
+    def get_name_from_email(self, user_email):
+        """
+        gets name from justice email as user.name is not guaranteed to be a name
+        (can be an email, forename only or handle)
+        """
+
+        name = user_email.split("@")[0]
+        forename, surname = name.split(".")
+        surname = "".join((c for c in surname if not c.isdigit()))
+        return forename, surname
+
+    def create_user(self, user_email):
+
+        if self.get_user_id(user_email):
+            log.debug("User already exists in Identity Store.")
+            return
+
+        try:
+            forename, surname = self.get_name_from_email(user_email)
+
+            self.client.create_user(
+                IdentityStoreId=self.sso_client.get_identity_store_id(),
+                UserName=user_email,
+                DisplayName=user_email,
+                Name={
+                    "FamilyName": surname,
+                    "GivenName": forename,
+                },
+                Emails=[{"Value": user_email, "Type": "EntraId", "Primary": True}],
+            )
+        except Exception as error:
+            log.exception(error)
+            raise error
+
+    def create_group_membership(self, group_name, user_email):
+
+        try:
+            membership_id = self.get_group_membership_id(group_name, user_email)
+
+            if membership_id is not None:
+                log.debug("User is already a member of this group. Skipping")
+                return
+
+            response = self.client.create_group_membership(
+                IdentityStoreId=self.sso_client.get_identity_store_id(),
+                GroupId=self.get_group_id(group_name),
+                MemberId={"UserId": self.get_user_id(user_email)},
+            )
+
+            return response
+        except Exception as error:
+            log.exception(error)
+            raise error
+
+    def delete_group_membership(self, group_name, user_email):
+        try:
+            membership_id = self.get_group_membership_id(group_name, user_email)
+
+            if membership_id is None:
+                log.debug("User is not a member of this group. Skipping")
+                return
+
+            self.client.delete_group_membership(
+                IdentityStoreId=self.sso_client.get_identity_store_id(),
+                MembershipId=membership_id,
+            )
+        except Exception as error:
+            log.exception(error.response["Error"]["Message"])
+            raise error
+
+    def add_user_to_group(self, justice_email, quicksight_group):
+        if not justice_email:
+            message = (
+                "Cannot create an Identity Center user without an associated @justice.gov.uk email"
+            )
+            log.exception(message)
+            raise Exception(message)
+
+        self.create_user(justice_email)
+        self.create_group_membership(quicksight_group, justice_email)
+        self.create_group_membership(settings.AZURE_HOLDING_GROUP_NAME, justice_email)

controlpanel/api/migrations/0055_alter_user_options.py

@@ -0,0 +1,29 @@
+# Generated by Django 5.1.2 on 2025-01-09 12:24
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("api", "0054_alter_tool_description"),
+    ]
+
+    operations = [
+        migrations.AlterModelOptions(
+            name="user",
+            options={
+                "ordering": ("username",),
+                "permissions": [
+                    (
+                        "quicksight_embed_author_access",
+                        "Can access embedded Quicksight as an author",
+                    ),
+                    (
+                        "quicksight_embed_reader_access",
+                        "Can access embedded Quicksight as a reader",
+                    ),
+                ],
+            },
+        ),
+    ]

controlpanel/api/models/__init__.py

@@ -15,6 +15,10 @@
 from controlpanel.api.models.s3bucket import S3Bucket
 from controlpanel.api.models.task import Task
 from controlpanel.api.models.tool import HomeDirectory, Tool, ToolDeployment
-from controlpanel.api.models.user import QUICKSIGHT_EMBED_PERMISSION, User
+from controlpanel.api.models.user import (
+    QUICKSIGHT_EMBED_AUTHOR_PERMISSION,
+    QUICKSIGHT_EMBED_READER_PERMISSION,
+    User,
+)
 from controlpanel.api.models.userapp import UserApp
 from controlpanel.api.models.users3bucket import UserS3Bucket

controlpanel/api/models/user.py

@@ -11,7 +11,8 @@
 from controlpanel.api.signals import prometheus_login_event
 from controlpanel.utils import sanitize_dns_label
 
-QUICKSIGHT_EMBED_PERMISSION = "quicksight_embed_access"
+QUICKSIGHT_EMBED_AUTHOR_PERMISSION = "quicksight_embed_author_access"
+QUICKSIGHT_EMBED_READER_PERMISSION = "quicksight_embed_reader_access"
 
 
 class User(AbstractUser):
@@ -50,7 +51,10 @@ class User(AbstractUser):
     class Meta:
         db_table = "control_panel_api_user"
         ordering = ("username",)
-        permissions = [(QUICKSIGHT_EMBED_PERMISSION, "Can access the embedded Quicksight")]
+        permissions = [
+            (QUICKSIGHT_EMBED_AUTHOR_PERMISSION, "Can access embedded Quicksight as an author"),
+            (QUICKSIGHT_EMBED_READER_PERMISSION, "Can access embedded Quicksight as a reader"),
+        ]
 
     def __repr__(self):
         return f"<User: {self.username} ({self.auth0_id})>"

controlpanel/frontend/forms.py

@@ -3,18 +3,21 @@
 
 # Third-party
 from django import forms
+from django.conf import settings
 from django.contrib.auth.models import Permission
 from django.contrib.postgres.forms import SimpleArrayField
 from django.core.exceptions import ValidationError
 from django.core.validators import RegexValidator, validate_email
 
 # First-party/Local
 from controlpanel.api import validators
+from controlpanel.api.aws import AWSIdentityStore
 from controlpanel.api.cluster import AWSRoleCategory
 from controlpanel.api.cluster import S3Folder as ClusterS3Folder
 from controlpanel.api.github import GithubAPI, RepositoryNotFound, extract_repo_info_from_url
 from controlpanel.api.models import (
-    QUICKSIGHT_EMBED_PERMISSION,
+    QUICKSIGHT_EMBED_AUTHOR_PERMISSION,
+    QUICKSIGHT_EMBED_READER_PERMISSION,
     App,
     Feedback,
     S3Bucket,
@@ -636,17 +639,30 @@ class CreateParameterForm(forms.Form):
 
 class QuicksightAccessForm(forms.Form):
     QUICKSIGHT_LEGACY = "quicksight_legacy"
-    QUICKSIGHT_COMPUTE = "quicksight_compute"
+    QUICKSIGHT_COMPUTE_AUTHOR = "quicksight_compute_author"
+    QUICKSIGHT_COMPUTE_READER = "quicksight_compute_reader"
 
     enable_quicksight = forms.MultipleChoiceField(
         choices=[
             (QUICKSIGHT_LEGACY, "Legacy"),
-            (QUICKSIGHT_COMPUTE, "Compute"),
+            (QUICKSIGHT_COMPUTE_AUTHOR, "Author"),
+            (QUICKSIGHT_COMPUTE_READER, "Reader"),
         ],
         widget=forms.CheckboxSelectMultiple,
         required=False,
     )
 
+    quicksight_config_data = {
+        QUICKSIGHT_COMPUTE_AUTHOR: {
+            "codename": QUICKSIGHT_EMBED_AUTHOR_PERMISSION,
+            "group": settings.QUICKSIGHT_AUTHOR_GROUP_NAME,
+        },
+        QUICKSIGHT_COMPUTE_READER: {
+            "codename": QUICKSIGHT_EMBED_READER_PERMISSION,
+            "group": settings.QUICKSIGHT_READER_GROUP_NAME,
+        },
+    }
+
     def __init__(self, *args, **kwargs):
         self.user = kwargs.pop("user")
         super().__init__(*args, **kwargs)
@@ -655,11 +671,28 @@ def grant_access(self):
         quicksight_access = self.cleaned_data["enable_quicksight"]
         self.user.set_quicksight_access(enable=self.QUICKSIGHT_LEGACY in quicksight_access)
 
-        permission = Permission.objects.get(codename=QUICKSIGHT_EMBED_PERMISSION)
-        if self.QUICKSIGHT_COMPUTE in quicksight_access:
+        self.set_quicksight_embed_access(self.QUICKSIGHT_COMPUTE_AUTHOR, quicksight_access)
+        self.set_quicksight_embed_access(self.QUICKSIGHT_COMPUTE_READER, quicksight_access)
+
+    def set_quicksight_embed_access(self, permission_name, quicksight_access):
+        identity_store = AWSIdentityStore(
+            settings.IDENTITY_CENTER_ASSUMED_ROLE,
+            "APCPIdentityCenterAccess",
+            settings.IDENTITY_CENTER_ACCOUNT_REGION,
+        )
+        if self.user.is_superuser:
+            return
+
+        codename = self.quicksight_config_data.get(permission_name)["codename"]
+        group = self.quicksight_config_data.get(permission_name)["group"]
+        permission = Permission.objects.get(codename=codename)
+
+        if permission_name in quicksight_access and not self.user.has_perm(f"api.{codename}"):
             self.user.user_permissions.add(permission)
-        else:
+            identity_store.add_user_to_group(self.user.justice_email, group)
+        elif self.user.has_perm(f"api.{codename}"):
             self.user.user_permissions.remove(permission)
+            identity_store.delete_group_membership(self.user.justice_email, group)
 
 
 class FeedbackForm(forms.ModelForm):

controlpanel/frontend/jinja2/user-detail.html

@@ -124,9 +124,15 @@ <h2 class="govuk-heading-m govuk-error-summary-heading">
           "checked": user.is_quicksight_enabled
         },
         {
-          "value": "quicksight_compute",
-          "text": "Embedded Quicksight enabled (running in the compute-" + env + " account)",
-          "checked": user.has_perm("api.quicksight_embed_access"),
+          "value": "quicksight_compute_author",
+          "text": "Embedded Quicksight enabled (running in the compute-" + env + " account) as an author",
+          "checked": user.has_perm("api.quicksight_embed_author_access"),
+          "disabled": user.is_superuser
+        },
+        {
+          "value": "quicksight_compute_reader",
+          "text": "Embedded Quicksight enabled (running in the compute-" + env + " account) as a reader",
+          "checked": user.has_perm("api.quicksight_embed_reader_access"),
           "disabled": user.is_superuser
         },
       ]

controlpanel/frontend/views/quicksight.py

@@ -16,7 +16,12 @@
 
 class QuicksightView(OIDCLoginRequiredMixin, PermissionRequiredMixin, TemplateView):
     template_name = "quicksight.html"
-    permission_required = "api.quicksight_embed_access"
+
+    def has_permission(self):
+        user = self.request.user
+        return user.has_perm("api.quicksight_embed_author_access") or user.has_perm(
+            "api.quicksight_embed_reader_access"
+        )
 
     def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
         context = super().get_context_data(**kwargs)

controlpanel/frontend/views/user.py

@@ -2,6 +2,7 @@
 from datetime import datetime, timedelta
 
 # Third-party
+from django.conf import settings
 from django.contrib import messages
 from django.contrib.auth.models import Permission
 from django.forms import BaseModelForm
@@ -16,8 +17,9 @@
 from rules.contrib.views import PermissionRequiredMixin
 
 # First-party/Local
+from controlpanel.api.aws import AWSIdentityStore
 from controlpanel.api.cluster import User as ClusterUser
-from controlpanel.api.models import User
+from controlpanel.api.models import QUICKSIGHT_EMBED_AUTHOR_PERMISSION, User
 from controlpanel.frontend import forms
 from controlpanel.frontend.mixins import PolicyAccessMixin
 from controlpanel.oidc import OIDCLoginRequiredMixin
@@ -98,6 +100,22 @@ class SetSuperadmin(OIDCLoginRequiredMixin, PermissionRequiredMixin, View):
     def post(self, request, *args, **kwargs):
         user = get_object_or_404(User, pk=kwargs["pk"])
         is_superuser = "is_superuser" in request.POST
+
+        identity_store = AWSIdentityStore(
+            settings.IDENTITY_CENTER_ASSUMED_ROLE,
+            "APCPIdentityCenterAccess",
+            settings.IDENTITY_CENTER_ACCOUNT_REGION,
+        )
+
+        if is_superuser:
+            identity_store.add_user_to_group(
+                user.justice_email, settings.QUICKSIGHT_AUTHOR_GROUP_NAME
+            )
+        else:
+            identity_store.delete_group_membership(
+                user.justice_email, settings.QUICKSIGHT_AUTHOR_GROUP_NAME
+            )
+
         user.is_superuser = is_superuser
         user.is_staff = is_superuser
         user.save()

controlpanel/settings/common.py

@@ -506,6 +506,13 @@
 QUICKSIGHT_ACCOUNT_REGION = os.environ.get("QUICKSIGHT_ACCOUNT_REGION")
 QUICKSIGHT_ASSUMED_ROLE = os.environ.get("QUICKSIGHT_ASSUMED_ROLE")
 
+IDENTITY_CENTER_ASSUMED_ROLE = os.environ.get("IDENTITY_CENTER_ASSUMED_ROLE")
+IDENTITY_CENTER_ACCOUNT_REGION = os.environ.get("IDENTITY_CENTER_ACCOUNT_REGION")
+QUICKSIGHT_READER_GROUP_NAME = os.environ.get("QUICKSIGHT_READER_GROUP_NAME")
+QUICKSIGHT_AUTHOR_GROUP_NAME = os.environ.get("QUICKSIGHT_AUTHOR_GROUP_NAME")
+QUICKSIGHT_ADMIN_GROUP_NAME = os.environ.get("QUICKSIGHT_ADMIN_GROUP_NAME")
+AZURE_HOLDING_GROUP_NAME = os.environ.get("AZURE_HOLDING_GROUP_NAME")
+
 # The EKS OIDC provider, referenced in user policies to allow service accounts
 # to grant AWS permissions.
 OIDC_EKS_PROVIDER = os.environ.get("OIDC_EKS_PROVIDER")