Add actions to manage users and domains

charm/charmcraft.yaml

@@ -50,3 +50,52 @@ config:
     git_ssh_key:
       type: string
       description: The private key for SSH authentication.
+
+actions:
+  create-user:
+      description: >
+        Create a user for the services that will be requesting the domains.
+        If it exists, the password will be updated.
+      properties:
+        username:
+          description: User name to be created.
+          type: string
+      required:
+        - username
+      required:
+        - username
+  allow-domains:
+      description: Grant user access to domains.
+      properties:
+        username:
+          description: User name to grant access for.
+          type: string
+        domains:
+          description: >
+            Comma separated list of domains to access access for.
+            The domains will be prefixed by '_acme-challenge.'
+      required:
+        - username
+        - domains
+  revoke-domains:
+      description: Revoke user access to domains.
+      properties:
+        username:
+          description: User name to revoke access for.
+          type: string
+        domains:
+          description: >
+            Comma separated list of domains to revoke access to.
+            The domains will be prefixed by '_acme-challenge.'
+          type: string
+      required:
+        - username
+        - domains
+  list-domains:
+      description: List the domains an user has access to.
+      properties:
+        username:
+          description: User name to query for.
+          type: string
+      required:
+        - username

httprequest_lego_provider/tests/__init__ .py → charm/src/__init__.py

@@ -1,3 +1,3 @@
 # Copyright 2024 Canonical Ltd.
 # See LICENSE file for licensing details.
-"""Unit tests for the HTTPRequest Lego Provider module."""
+"""Charm module."""

charm/src/actions.py

@@ -0,0 +1,108 @@
+# Copyright 2024 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+"""HTTPRequest LEGO provider charm actions."""
+
+# pylint: disable=protected-access
+
+import logging
+import secrets
+
+import ops
+import xiilib.django
+
+logger = logging.getLogger(__name__)
+
+
+class NotReadyError(Exception):
+    """Exception thrown when needed resources are not ready."""
+
+
+class Observer(ops.Object):
+    """Charm actions observer."""
+
+    def __init__(self, charm: xiilib.django.Charm):
+        """Initialize the observer and register actions handlers.
+
+        Args:
+            charm: The parent charm to attach the observer to.
+        """
+        super().__init__(charm, "actions-observer")
+        self.charm = charm
+
+        charm.framework.observe(charm.on.create_user_action, self._create_user_action)
+        charm.framework.observe(charm.on.allow_domains_action, self._allow_domains)
+        charm.framework.observe(charm.on.revoke_domains_action, self._revoke_domains)
+        charm.framework.observe(charm.on.list_domains_action, self._list_domains)
+
+    def _generate_password(self) -> str:
+        """Generate a new password.
+
+        Returns: the new password.
+        """
+        return secrets.token_urlsafe(30)
+
+    def _execute_command(self, command: list[str], event: ops.ActionEvent) -> None:
+        """Prepare the scripts for exxecution.
+
+        Args:
+            command: the management command to execute.
+            event: the event triggering the original action.
+
+        Raises:
+            ExecError: if an error occurs while executing the script
+        """
+        container = self.charm.unit.get_container(self.charm._CONTAINER_NAME)
+        if not container.can_connect() or not self.charm._databases.is_ready():
+            event.fail("Service not yet ready.")
+
+        process = container.exec(
+            ["python3", "manage.py"] + command,
+            working_dir=str(self.charm._BASE_DIR / "app"),
+            environment=self.charm.gen_env(),
+        )
+        try:
+            stdout, _ = process.wait_output()
+            event.set_results({"result": stdout})
+        except ops.pebble.ExecError as ex:
+            logger.exception("Action %s failed: %s %s", ex.command, ex.stdout, ex.stderr)
+            event.fail(f"Failed: {ex.stderr!r}")
+
+    def _create_user_action(self, event: ops.ActionEvent) -> None:
+        """Handle create-user and update-password actions.
+
+        Args:
+            event: The event fired by the action.
+        """
+        username = event.params["username"]
+        password = self._generate_password()
+        self._execute_command(["create_user", username, password], event)
+
+    def _allow_domains(self, event: ops.ActionEvent) -> None:
+        """Handle the allow-domains action.
+
+        Args:
+            event: The event fired by the action.
+        """
+        username = event.params["username"]
+        domains = event.params["domains"].split(",")
+        self._execute_command(["allow_domains", username] + domains, event)
+
+    def _revoke_domains(self, event: ops.ActionEvent) -> None:
+        """Handle the allow-domains action.
+
+        Args:
+            event: The event fired by the action.
+        """
+        username = event.params["username"]
+        domains = event.params["domains"].split(",")
+        self._execute_command(["revoke_domains", username] + domains, event)
+
+    def _list_domains(self, event: ops.ActionEvent) -> None:
+        """Handle the allow-domains action.
+
+        Args:
+            event: The event fired by the action.
+        """
+        username = event.params["username"]
+        self._execute_command(["list_domains", username], event)

charm/src/charm.py

@@ -7,8 +7,8 @@
 import logging
 import typing
 
+import actions
 import ops
-
 import xiilib.django
 
 logger = logging.getLogger(__name__)
@@ -19,6 +19,7 @@
 KNOWN_HOSTS_PATH = "/var/lib/pebble/default/.ssh/known_hosts"
 RSA_PATH = "/var/lib/pebble/default/.ssh/id_rsa"
 
+
 class DjangoCharm(xiilib.django.Charm):
     """Flask Charm service."""
 
@@ -29,10 +30,11 @@ def __init__(self, *args: typing.Any) -> None:
             args: passthrough to CharmBase.
         """
         super().__init__(*args)
+        self.actions_observer = actions.Observer(self)
         self.framework.observe(self.on.collect_app_status, self._on_collect_app_status)
 
     def _on_config_changed(self, _event: ops.ConfigChangedEvent) -> None:
-        """"Config changed handler.
+        """Config changed handler.
 
         Args:
             event: the event triggering the handler.
@@ -41,7 +43,7 @@ def _on_config_changed(self, _event: ops.ConfigChangedEvent) -> None:
         super()._on_config_changed(_event)
 
     def _on_django_app_pebble_ready(self, _event: ops.PebbleReadyEvent) -> None:
-        """"Pebble ready handler.
+        """Pebble ready handler.
 
         Args:
             event: the event triggering the handler.
@@ -77,9 +79,9 @@ def _copy_files(self) -> None:
             group=DJANGO_GROUP,
             permissions=0o600,
         )
-    
+
     def _on_collect_app_status(self, _: ops.CollectStatusEvent) -> None:
-        """"Handle the status changes.
+        """Handle the status changes.
 
         Args:
             event: the event triggering the handler.
@@ -92,5 +94,5 @@ def _on_collect_app_status(self, _: ops.CollectStatusEvent) -> None:
             return
 
 
-if __name__ == "__main__":
+if __name__ == "__main__":  # pragma: no cover
     ops.main.main(DjangoCharm)

httprequest_lego_provider/management/__init__.py

@@ -0,0 +1,3 @@
+# Copyright 2024 Canonical Ltd.
+# See LICENSE file for licensing details.
+"""HTTPRequest management."""

httprequest_lego_provider/management/commands/__init__.py

@@ -0,0 +1,3 @@
+# Copyright 2024 Canonical Ltd.
+# See LICENSE file for licensing details.
+"""HTTPRequest management commands."""

httprequest_lego_provider/management/commands/allow_domains.py

@@ -0,0 +1,58 @@
+# Copyright 2024 Canonical Ltd.
+# See LICENSE file for licensing details.
+"""Allow domains module."""
+
+# imported-auth-user has to be disable as the conflicting import is needed for typing
+# pylint:disable=duplicate-code,imported-auth-user
+
+from django.contrib.auth.models import User
+from django.core.management.base import BaseCommand, CommandError
+
+from httprequest_lego_provider.forms import FQDN_PREFIX
+from httprequest_lego_provider.models import Domain, DomainUserPermission
+
+
+class Command(BaseCommand):
+    """Command to grant access to domains to a user.
+
+    Attrs:
+        help: help message to display.
+    """
+
+    help = "Grant user access to domains."
+
+    def add_arguments(self, parser):
+        """Argument parser.
+
+        Args:
+            parser: the cmd line parser.
+        """
+        parser.add_argument("username", nargs=None, type=str)
+        parser.add_argument("domains", nargs="+", type=str)
+
+    def handle(self, *args, **options):
+        """Command handler.
+
+        Args:
+            args: args.
+            options: options.
+
+        Raises:
+            CommandError: if the user is not found.
+        """
+        username = options["username"]
+        domains = options["domains"]
+        try:
+            user = User.objects.get(username=username)
+        except User.DoesNotExist as exc:
+            raise CommandError(f'User "{username}" does not exist') from exc
+        for domain_name in domains:
+            fqdn = (
+                domain_name
+                if domain_name.startswith(FQDN_PREFIX)
+                else f"{FQDN_PREFIX}{domain_name}"
+            )
+            domain, _ = Domain.objects.get_or_create(fqdn=fqdn)
+            DomainUserPermission.objects.get_or_create(domain=domain, user=user)
+
+        self.stdout.write(self.style.SUCCESS(f'Granted "{", ".join(domains)}" for "{username}"'))

httprequest_lego_provider/management/commands/create_user.py

@@ -0,0 +1,44 @@
+# Copyright 2024 Canonical Ltd.
+# See LICENSE file for licensing details.
+"""Create user module."""
+
+# imported-auth-user has to be disable as the conflicting import is needed for typing
+# pylint:disable=imported-auth-user
+
+from django.contrib.auth.models import User
+from django.core.management.base import BaseCommand
+
+
+class Command(BaseCommand):
+    """Command for user creation.
+
+    Attrs:
+        help: help message to display.
+    """
+
+    help = "Create a user or update its password."
+
+    def add_arguments(self, parser):
+        """Argument parser.
+
+        Args:
+            parser: the cmd line parser.
+        """
+        parser.add_argument("username", nargs=None, type=str)
+        parser.add_argument("password", nargs=None, type=str)
+
+    def handle(self, *args, **options):
+        """Command handler.
+
+        Args:
+            args: args.
+            options: options.
+        """
+        username = options["username"]
+        password = options["password"]
+        user, _ = User.objects.get_or_create(username=username, defaults={"password": password})
+        user.save()
+
+        self.stdout.write(
+            self.style.SUCCESS(f'Created or updated "{username}" with password "{password}"')
+        )

httprequest_lego_provider/management/commands/list_domains.py

@@ -0,0 +1,48 @@
+# Copyright 2024 Canonical Ltd.
+# See LICENSE file for licensing details.
+"""List domains module."""
+
+# imported-auth-user has to be disable as the conflicting import is needed for typing
+# pylint:disable=imported-auth-user
+
+from django.contrib.auth.models import User
+from django.core.management.base import BaseCommand, CommandError
+
+from httprequest_lego_provider.models import DomainUserPermission
+
+
+class Command(BaseCommand):
+    """Command to list the domains an user has access to.
+
+    Attrs:
+        help: help message to display.
+    """
+
+    help = "Create a user or update its password."
+
+    def add_arguments(self, parser):
+        """Argument parser.
+
+        Args:
+            parser: the cmd line parser.
+        """
+        parser.add_argument("username", nargs=None, type=str)
+
+    def handle(self, *args, **options):
+        """Command handler.
+
+        Args:
+            args: args.
+            options: options.
+
+        Raises:
+            CommandError: if the user is not found.
+        """
+        username = options["username"]
+        try:
+            user = User.objects.get(username=username)
+        except User.DoesNotExist as exc:
+            raise CommandError(f'User "{username}" does not exist') from exc
+        dups = DomainUserPermission.objects.filter(user=user)
+
+        self.stdout.write(self.style.SUCCESS(", ".join([dup.domain.fqdn for dup in dups])))