[adminapi] add an endpoint to register a user

app/adminapi/command.py

@@ -66,6 +66,7 @@ def prepare(self):
             layer2_repo=repositories.Layer2Repository(self.pg_storage, log),
             metadata_repo=repositories.MetadataRepository(self.pg_storage),
             authenticator=authenticator,
+            storage=self.pg_storage,
             clients=clients.Clients(cfg.clients.ads_token),
             table_stats_cache=self.table_stats_cache,
         )

app/adminapi/domain/__init__.py

@@ -1,6 +1,6 @@
 from app.adminapi.domain.actions import Actions
+from app.adminapi.domain.auth import AuthManager
 from app.adminapi.domain.crossmatch import CrossmatchManager
-from app.adminapi.domain.login import LoginManager
 from app.adminapi.domain.mock import get_mock_actions
 from app.adminapi.domain.sources import SourceManager
 from app.adminapi.domain.table_upload import TableUploadManager
@@ -9,7 +9,7 @@
     "Actions",
     "CrossmatchManager",
     "get_mock_actions",
-    "LoginManager",
+    "AuthManager",
     "TableUploadManager",
     "SourceManager",
 ]

app/adminapi/domain/actions.py

@@ -2,9 +2,11 @@
 
 from app.adminapi import cache, clients
 from app.adminapi import presentation as adminapi
-from app.adminapi.domain import catalogs, crossmatch, layer1_write, login, pgc, sources, table_upload
+from app.adminapi.domain import auth as admin_auth
+from app.adminapi.domain import catalogs, crossmatch, layer1_write, pgc, sources, table_upload
 from app.data import repositories
 from app.lib import auth
+from app.lib.storage import postgres
 from app.lib.tap import types as tap_types
 
 _ADMIN_TAP_SYNC_QUERY_TIMEOUT_SECONDS = 60
@@ -26,12 +28,13 @@ def __init__(
         layer2_repo: repositories.Layer2Repository,
         metadata_repo: repositories.MetadataRepository,
         authenticator: auth.Authenticator,
+        storage: postgres.PgStorage,
         clients: clients.Clients,
         table_stats_cache: cache.BackgroundCache[adminapi.TableStatsSnapshot],
     ):
         self.metadata_repo = metadata_repo
         self.source_manager = sources.SourceManager(common_repo)
-        self.login_manager = login.LoginManager(authenticator)
+        self.auth_manager = admin_auth.AuthManager(authenticator, storage)
         self.table_upload_manager = table_upload.TableUploadManager(
             common_repo,
             layer0_repo,
@@ -48,10 +51,13 @@ def create_source(self, r: adminapi.CreateSourceRequest) -> adminapi.CreateSourc
         return self.source_manager.create_source(r)
 
     def login(self, r: adminapi.LoginRequest) -> adminapi.LoginResponse:
-        return self.login_manager.login(r)
+        return self.auth_manager.login(r)
 
     def logout(self, token: str) -> adminapi.LogoutResponse:
-        return self.login_manager.logout(token)
+        return self.auth_manager.logout(token)
+
+    def register(self, r: adminapi.RegisterRequest) -> adminapi.RegisterResponse:
+        return self.auth_manager.register(r)
 
     def add_data(self, r: adminapi.AddDataRequest) -> adminapi.AddDataResponse:
         return self.table_upload_manager.add_data(r)

app/adminapi/domain/auth.py

@@ -0,0 +1,62 @@
+import bcrypt
+from psycopg import sql
+
+from app.adminapi import presentation as adminapi
+from app.lib import auth
+from app.lib.storage import postgres
+from app.lib.web.errors import ConflictError, UnauthorizedError
+
+
+class AuthManager(postgres.TransactionalPGRepository):
+    def __init__(self, authenticator: auth.Authenticator, storage: postgres.PgStorage) -> None:
+        super().__init__(storage)
+        self.authenticator = authenticator
+
+    def login(self, r: adminapi.LoginRequest) -> adminapi.LoginResponse:
+        token, is_authenticated = self.authenticator.login(r.username, r.password)
+
+        if not is_authenticated:
+            raise UnauthorizedError("invalid username or password")
+
+        return adminapi.LoginResponse(token=token)
+
+    def logout(self, token: str) -> adminapi.LogoutResponse:
+        self.authenticator.revoke(token)
+        return adminapi.LogoutResponse()
+
+    def register(self, r: adminapi.RegisterRequest) -> adminapi.RegisterResponse:
+        self._ensure_available(r.username, r.email)
+
+        password_hash = bcrypt.hashpw(r.password.encode(), bcrypt.gensalt())
+
+        with self.with_tx():
+            self._storage.exec(
+                "INSERT INTO private.users (login, name, email, role, password_hash) VALUES (%s, %s, %s, 'admin', %s)",
+                params=[r.username, r.username, r.email, password_hash],
+            )
+            self._storage.exec(
+                sql.SQL("CREATE ROLE {} LOGIN PASSWORD {}").format(
+                    sql.Identifier(r.username),
+                    sql.Literal(r.password),
+                )
+            )
+            self._storage.exec(sql.SQL("GRANT db_reader TO {}").format(sql.Identifier(r.username)))
+
+        return adminapi.RegisterResponse()
+
+    def _ensure_available(self, username: str, email: str) -> None:
+        existing = self._storage.query(
+            "SELECT login, email FROM private.users WHERE login = %s OR email = %s",
+            params=[username, email],
+        )
+        if existing:
+            if existing[0]["login"] == username:
+                raise ConflictError(f"user '{username}' already exists")
+            raise ConflictError(f"email '{email}' already exists")
+
+        role_exists = self._storage.query(
+            "SELECT 1 FROM pg_roles WHERE rolname = %s",
+            params=[username],
+        )
+        if role_exists:
+            raise ConflictError(f"database user '{username}' already exists")

app/adminapi/domain/mock.py

@@ -24,6 +24,7 @@ def get_mock_actions():
         layer2_repo=mock.MagicMock(),
         metadata_repo=mock.MagicMock(),
         authenticator=auth.NoopAuthenticator(),
+        storage=mock.MagicMock(),
         clients=clients.get_mock_clients(),
         table_stats_cache=get_mock_table_stats_cache(),
     )

app/adminapi/presentation/interface.py

@@ -216,6 +216,16 @@ class LogoutResponse(pydantic.BaseModel):
     pass
 
 
+class RegisterRequest(pydantic.BaseModel):
+    username: str
+    email: str
+    password: str
+
+
+class RegisterResponse(pydantic.BaseModel):
+    pass
+
+
 class RecordCrossmatchMetadata(pydantic.BaseModel):
     possible_matches: list[int] | None = None
     pgc: int | None = None
@@ -486,6 +496,10 @@ def login(self, r: LoginRequest) -> LoginResponse:
     def logout(self, token: str) -> LogoutResponse:
         pass
 
+    @abc.abstractmethod
+    def register(self, r: RegisterRequest) -> RegisterResponse:
+        pass
+
     @abc.abstractmethod
     def get_records(self, r: GetRecordsRequest) -> GetRecordsResponse:
         pass

app/adminapi/presentation/server.py

@@ -89,6 +89,15 @@ def logout(
     ) -> server.APIOkResponse[interface.LogoutResponse]:
         return server.APIOkResponse(data=self.actions.logout(_logout_token(request)))
 
+    def register(
+        self,
+        request: fastapi.Request,
+        body: interface.RegisterRequest,
+    ) -> server.APIOkResponse[interface.RegisterResponse]:
+        _ = request
+        response = self.actions.register(body)
+        return server.APIOkResponse(data=response)
+
     def get_record_crossmatch(
         self, request: Annotated[interface.GetRecordCrossmatchRequest, fastapi.Query()]
     ) -> server.APIOkResponse[interface.GetRecordCrossmatchResponse]:
@@ -292,6 +301,17 @@ def __init__(
                 allowed_roles=admin_only,
                 rate_limit="10/minute",
             ),
+            server.Route(
+                "/v1/register",
+                http.HTTPMethod.POST,
+                api.register,
+                "Register user",
+                "Creates a backend user and a database reader role.",
+                allowed_roles=admin_only,
+                rate_limit="10/minute",
+                log_request_body=False,
+                audit_action=True,
+            ),
             server.Route(
                 "/v1/records/crossmatch",
                 http.HTTPMethod.POST,

tests/adminapi/integration/adminapi_auth_test.py

@@ -1,11 +1,16 @@
+import hashlib
 import os
+import secrets
 import subprocess
 import time
 import unittest
 from concurrent import futures
+from datetime import UTC, datetime, timedelta
 
+import bcrypt
 import requests
 import structlog
+from psycopg import sql
 
 from app.lib import audit
 from tests import lib
@@ -118,6 +123,23 @@ def _login_and_get_token(self) -> str:
         self.assertEqual(r.status_code, 200)
         return r.json()["data"]["token"]
 
+    def _seed_bearer_token(self, login: str) -> str:
+        storage = self.pg_storage.get_storage()
+        user = storage.query_one(
+            "SELECT id FROM private.users WHERE login = %s",
+            params=[login],
+        )
+        token = secrets.token_hex(16)
+        storage.exec(
+            "INSERT INTO private.tokens (token_hash, user_id, expiry_time) VALUES (%s, %s, %s)",
+            params=[
+                hashlib.sha256(token.encode()).digest(),
+                user["id"],
+                datetime.now(UTC) + timedelta(days=14),
+            ],
+        )
+        return token
+
     def _assert_token_works(self, token: str) -> None:
         r = requests.post(
             f"{self.base}/v1/source",
@@ -273,3 +295,99 @@ def test_logout_revokes_token(self):
         self.assertEqual(r_out.status_code, 200)
 
         self._assert_token_rejected(token)
+
+    _registered_login = "integration_registered_user"
+    _registered_email = "integration_registered_user@example.com"
+    _registered_password = "registered-user-secret"
+
+    def _cleanup_registered_user(self) -> None:
+        storage = self.pg_storage.get_storage()
+        storage.exec(
+            "DELETE FROM private.tokens WHERE user_id IN (SELECT id FROM private.users WHERE login = %s)",
+            params=[self._registered_login],
+        )
+        storage.exec("DELETE FROM private.users WHERE login = %s", params=[self._registered_login])
+        if storage.query("SELECT 1 FROM pg_roles WHERE rolname = %s", params=[self._registered_login]):
+            storage.exec(sql.SQL("DROP ROLE {}").format(sql.Identifier(self._registered_login)))
+
+    def test_register(self) -> None:
+        r = requests.post(
+            f"{self.base}/v1/register",
+            json={
+                "username": self._registered_login,
+                "email": self._registered_email,
+                "password": self._registered_password,
+            },
+            timeout=5,
+        )
+        self.assertEqual(r.status_code, 401)
+        self.assertEqual(r.json()["message"], "No authorization header")
+
+        self._cleanup_registered_user()
+        try:
+            admin_token = self._seed_bearer_token(self._login)
+            body = {
+                "username": self._registered_login,
+                "email": self._registered_email,
+                "password": self._registered_password,
+            }
+
+            created = requests.post(
+                f"{self.base}/v1/register",
+                headers={"Authorization": f"Bearer {admin_token}"},
+                json=body,
+                timeout=5,
+            )
+            self.assertEqual(created.status_code, 200)
+            self.assertEqual(created.json()["data"], {})
+
+            storage = self.pg_storage.get_storage()
+            user = storage.query_one(
+                "SELECT login, email, password_hash FROM private.users WHERE login = %s",
+                params=[self._registered_login],
+            )
+            self.assertEqual(user["login"], self._registered_login)
+            self.assertEqual(user["email"], self._registered_email)
+            self.assertTrue(bcrypt.checkpw(self._registered_password.encode(), user["password_hash"]))
+
+            role = storage.query_one(
+                """
+                SELECT r.rolname
+                FROM pg_roles AS r
+                JOIN pg_auth_members AS am ON r.oid = am.member
+                JOIN pg_roles AS m ON am.roleid = m.oid
+                WHERE r.rolname = %s AND m.rolname = 'db_reader'
+                """,
+                params=[self._registered_login],
+            )
+            self.assertEqual(role["rolname"], self._registered_login)
+
+            tokens_before = storage.query(
+                "SELECT token_hash FROM private.tokens WHERE user_id = (SELECT id FROM private.users WHERE login = %s)",
+                params=[self._registered_login],
+            )
+            self.assertEqual(tokens_before, [])
+
+            duplicate_username = requests.post(
+                f"{self.base}/v1/register",
+                headers={"Authorization": f"Bearer {admin_token}"},
+                json=body,
+                timeout=5,
+            )
+            self.assertEqual(duplicate_username.status_code, 409)
+            self.assertIn("already exists", duplicate_username.json()["message"])
+
+            duplicate_email = requests.post(
+                f"{self.base}/v1/register",
+                headers={"Authorization": f"Bearer {admin_token}"},
+                json={
+                    "username": "another_registered_user",
+                    "email": self._registered_email,
+                    "password": self._registered_password,
+                },
+                timeout=5,
+            )
+            self.assertEqual(duplicate_email.status_code, 409)
+            self.assertIn("already exists", duplicate_email.json()["message"])
+        finally:
+            self._cleanup_registered_user()

tests/adminapi/integration/catalogs_api_test.py

@@ -28,6 +28,7 @@ def setUp(self) -> None:
             layer2_repo=repositories.Layer2Repository(pg, log),
             metadata_repo=repositories.MetadataRepository(pg),
             authenticator=auth.NoopAuthenticator(),
+            storage=pg,
             clients=clients.Clients(ads_token="test"),
             table_stats_cache=get_mock_table_stats_cache(),
         )

tests/adminapi/integration/tap_sync_test.py

@@ -29,6 +29,7 @@ def setUp(self) -> None:
             layer2_repo=repositories.Layer2Repository(pg, log),
             metadata_repo=repositories.MetadataRepository(pg),
             authenticator=auth.NoopAuthenticator(),
+            storage=pg,
             clients=clients.Clients(ads_token="test"),
             table_stats_cache=get_mock_table_stats_cache(),
         )