PB-1044 Add interface for cognito

.env.default

@@ -7,5 +7,15 @@ DB_HOST=localhost
 DB_NAME_TEST=postgres_test
 DJANGO_SETTINGS_MODULE=config.settings_dev
 SECRET_KEY=django-insecure-6-72r#zx=sv6v@-4k@uf1gv32me@%yr*oqa*fu8&5l&a!ws)5#
+COGNITO_POOL_ID=local_PoolPrty
+COGNITO_PORT=9229
+COGNITO_ENDPOINT_URL=http://localhost:9229
+COGNITO_FLAG_NAME=dev:custom:managed_by_service
+
+# used for local development
+AWS_REGION=eu-central-2
+AWS_DEFAULT_REGION=eu-central-2
+AWS_ACCESS_KEY_ID=123
+AWS_SECRET_ACCESS_KEY=123
 
 DEBUG=True

.gitignore

@@ -7,3 +7,6 @@ var/
 
 # IDE config files
 .vscode
+
+# Local database
+.volumes

Pipfile

@@ -12,6 +12,7 @@ gunicorn = "~=23.0"
 pyyaml = "~=6.0.2"
 gevent = "~=24.2"
 logging-utilities = "~=4.4.1"
+boto3 = "~=1.35"
 
 [dev-packages]
 yapf = "*"
@@ -30,6 +31,7 @@ mypy = "*"
 types-gevent = "*"
 django-stubs = "~=5.0.4"
 debugpy = "*"
+boto3-stubs = {extras = ["cognito-idp"], version = "~=1.35"}
 
 [requires]
 python_version = "3.12"

README.md

@@ -55,6 +55,28 @@ You may want to do an initial sync of your database by applying the most recent
 app/manage.py migrate
 ```
 
+## Cognito
+
+This project uses Amazon Cognito user identity and access management. It uses a custom user attribute to
+mark users managed_by_service by this service.
+
+To synchronize all local users with cognito, run:
+
+```bash
+app/manage.py cognito_sync
+```
+
+### Local Cognito
+
+For local testing the connection to cognito, [cognito-local](https://github.com/jagregory/cognito-local) is used.
+`cognito-local` stores all of its data as simple JSON files in its volume (`.volumes/cognito/db/`).
+
+You can also use the AWS CLI together with `cognito-local` by specifying the local endpoint, for example:
+
+```bash
+aws --endpoint $COGNITO_ENDPOINT_URL cognito-idp list-users --user-pool-id $COGNITO_POOL_ID
+```
+
 ## Local Development
 
 ### vs code Integration

app/cognito/apps.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class CognitoConfig(AppConfig):
+    default_auto_field = 'django.db.models.BigAutoField'
+    name = 'cognito'

app/cognito/management/commands/cognito_sync.py

@@ -0,0 +1,141 @@
+from typing import TYPE_CHECKING
+from typing import Any
+
+from access.models import User
+from cognito.utils.client import Client
+from cognito.utils.client import user_attributes_to_dict
+from utils.command import CommandHandler
+from utils.command import CustomBaseCommand
+
+from django.core.management.base import CommandParser
+
+if TYPE_CHECKING:
+    from mypy_boto3_cognito_idp.type_defs import UserTypeTypeDef
+
+
+class Handler(CommandHandler):
+
+    def __init__(self, command: CustomBaseCommand, options: dict[str, Any]) -> None:
+        super().__init__(command, options)
+        self.clear = options['clear']
+        self.dry_run = options['dry_run']
+        self.client = Client()
+        self.counts = {'added': 0, 'deleted': 0, 'updated': 0}
+
+    def clear_users(self) -> None:
+        """ Remove all existing cognito users. """
+
+        for user in self.client.list_users():
+            self.counts['deleted'] += 1
+            username = user['Username']
+            self.print(f'deleting user {username}')
+            if not self.dry_run:
+                deleted = self.client.delete_user(username)
+                if not deleted:
+                    self.print_error(
+                        'Could not delete %s, might not exist or might be unmanged', username
+                    )
+
+    def add_user(self, user: User) -> None:
+        """ Add a local user to cognito. """
+
+        self.counts['added'] += 1
+        self.print(f'adding user {user.username}')
+        if not self.dry_run:
+            created = self.client.create_user(user.username, user.email)
+            if not created:
+                self.print_error(
+                    'Could not create %s, might already exist as unmanaged user', user.username
+                )
+
+    def delete_user(self, username: str) -> None:
+        """ Delete a remote user from cognito. """
+
+        self.counts['deleted'] += 1
+        self.print(f'deleting user {username}')
+        if not self.dry_run:
+            deleted = self.client.delete_user(username)
+            if not deleted:
+                self.print_error(
+                    'Could not delete %s, might not exist or might be unmanaged', username
+                )
+
+    def update_user(self, local_user: User, remote_user: 'UserTypeTypeDef') -> None:
+        """ Update a remote user in cognito. """
+
+        remote_attributes = user_attributes_to_dict(remote_user['Attributes'])
+        if local_user.email != remote_attributes.get('email'):
+            self.counts['updated'] += 1
+            self.print(f'updating user {local_user.username}')
+            if not self.dry_run:
+                updated = self.client.update_user(local_user.username, local_user.email)
+                if not updated:
+                    self.print_error(
+                        'Could not update %s, might not exist or might be unmanaged',
+                        local_user.username
+                    )
+
+    def sync_users(self) -> None:
+        """ Synchronizes local and cognito users. """
+
+        # Get all remote and local users
+        local_users = {user.username: user for user in User.objects.all()}
+        local_usernames = set(local_users.keys())
+        remote_users = {user['Username']: user for user in self.client.list_users()}
+        remote_usernames = set(remote_users.keys())
+
+        for username in local_usernames.difference(remote_usernames):
+            self.add_user(local_users[username])
+
+        for username in remote_usernames.difference(local_usernames):
+            self.delete_user(username)
+
+        for username in local_usernames.intersection(remote_usernames):
+            self.update_user(local_users[username], remote_users[username])
+
+    def run(self) -> None:
+        """ Main entry point of command. """
+
+        # Clear data
+        if self.clear:
+            self.print_warning('This action will delete all managed users from cognito', level=0)
+            confirm = input('are you sure you want to proceed? [yes/no]: ')
+            if confirm.lower() != 'yes':
+                self.print_warning('operation cancelled', level=0)
+                return
+
+            self.clear_users()
+
+        self.sync_users()
+
+        # Print counts
+        printed = False
+        for operation, count in self.counts.items():
+            if count:
+                printed = True
+                self.print_success(f'{count} user(s) {operation}')
+        if not printed:
+            self.print_success('nothing to be done')
+
+        if self.dry_run:
+            self.print_warning('dry run, nothing has been done')
+
+
+class Command(CustomBaseCommand):
+    help = "Synchronizes local users with cognito"
+
+    def add_arguments(self, parser: CommandParser) -> None:
+        super().add_arguments(parser)
+        parser.add_argument(
+            '--clear',
+            action='store_true',
+            help='Delete existing users in cognito before synchronizing',
+        )
+        parser.add_argument(
+            '--dry-run',
+            action='store_true',
+            help='Dry run, abort transaction in the end',
+        )
+
+    def handle(self, *args: Any, **options: Any) -> None:
+        Handler(self, options).run()