Change how services are provided so that they don't go via "extensions"

config.example.json

@@ -11,5 +11,8 @@
   "database": {
     "url": "sqlite://",
     "echo": false
+  },
+  "auth": {
+    "ed25519_private_key": "A base64 encoded EdDSA Private Key"
   }
 }

derivation.nix

@@ -3,7 +3,7 @@ with python313Packages;
 buildPythonApplication {
   pname = "cereal";
   version = "0.0.1";
-  propagatedBuildInputs = [ flask requests waitress sqlalchemy argon2-cffi];
+  propagatedBuildInputs = [ flask requests waitress sqlalchemy argon2-cffi pyjwt cryptography];
   src = ./.;
   pyproject = true;
   build-system = [setuptools];

shell.nix

@@ -6,6 +6,8 @@ let
     waitress
     sqlalchemy
     argon2-cffi
+    pyjwt
+    cryptography
   ]);
 in
 with pkgs;

src/config/__init__.py

@@ -4,6 +4,7 @@ from typing import Any
 
 from .logging import Logging
 from .email import Email
+from .auth import Auth
 from .database import Database
 from .parse import assert_key_of_type, parse_nested_config
 
@@ -15,19 +16,25 @@ class Config:
     logging: Logging
     email: Email
     database: Database
+    auth: Auth
 
     @classmethod
     def from_dict(cls, config: dict[str, Any]) -> 'Config':
-        assert_key_of_type(config, 'logging', dict)
-        assert_key_of_type(config, 'email', dict)
         assert_key_of_type(config, 'host', str)
         assert_key_of_type(config, 'port', int)
+
         log_config = parse_nested_config(config, 'logging', Logging.from_dict)
         email_config = parse_nested_config(config, 'email', Email.from_dict)
         db_config = parse_nested_config(config, 'database', Database.from_dict)
+        auth_config = parse_nested_config(config, 'auth', Auth.from_dict)
 
         return Config(
-            host=config['host'], port=config['port'], email=email_config, logging=log_config, database=db_config
+            host=config['host'],
+            port=config['port'],
+            email=email_config,
+            logging=log_config,
+            database=db_config,
+            auth=auth_config,
         )
 
 

src/config/auth.py

@@ -0,0 +1,16 @@
+from base64 import b64decode
+from typing import Any
+from src.config.parse import assert_key_of_type
+from src.utils.secret import SecretBox
+from dataclasses import dataclass
+
+
+@dataclass
+class Auth:
+    ed25519_private_key: SecretBox[bytes]
+
+    @classmethod
+    def from_dict(cls, config: dict[str, Any]) -> 'Auth':
+        assert_key_of_type(config, 'ed25519_private_key', str)
+        private_key = b64decode(config['ed25519_private_key'])
+        return Auth(ed25519_private_key=SecretBox(private_key))

src/config/logging.py

@@ -2,7 +2,7 @@ from typing import Any
 from dataclasses import dataclass
 import logging
 
-from .parse import assert_key_of_type, ConfigParseError
+from .parse import assert_key_of_type, ParseError
 
 LOG_LEVELS = {
     'critical': logging.CRITICAL,
@@ -25,7 +25,7 @@ class Logging:
         assert_key_of_type(config, 'level', str)
         log_level_setting = config['level']
         if log_level_setting not in LOG_LEVELS:
-            raise ConfigParseError(
+            raise ParseError(
                 ['level'], f'unknown log level "{log_level_setting}". Expected one of: {", ".join(LOG_LEVELS.keys())}'
             )
 

src/config/parse.py

@@ -1,7 +1,7 @@
 from typing import Any, TypeVar, Callable
 
 
-class ConfigParseError(Exception):
+class ParseError(Exception):
     """An error for when parsing a config fails"""
 
     def __init__(self, keypath: list[str], issue: str):
@@ -14,10 +14,10 @@ class ConfigParseError(Exception):
 
 def assert_key_of_type(config: dict[str, Any], key: str, kind: Any):
     if key not in config:
-        raise ConfigParseError([key], 'missing')
+        raise ParseError([key], 'missing')
 
     if not isinstance(config[key], kind):
-        raise ConfigParseError([key], f'type of "{key}" was {type(config[key])}, expected {kind}')
+        raise ParseError([key], f'type of "{key}" was {type(config[key])}, expected {kind}')
 
 
 T = TypeVar('T')
@@ -27,5 +27,5 @@ def parse_nested_config(config: dict[str, Any], key: str, parse: Callable[[dict[
     assert_key_of_type(config, key, dict)
     try:
         return parse(config[key])
-    except ConfigParseError as e:
+    except ParseError as e:
-        raise ConfigParseError([key, *e.keypath], e.issue)
+        raise ParseError([key, *e.keypath], e.issue)

src/main.py

@@ -1,8 +1,10 @@
 #!/usr/bin/env python
 
-from src.notifications.service import make_notification_service
+from src.services.notifications import NotificationService, get_notification_service
-from src.services.email import get_service as get_email_service
+from src.services.email import EmailService, get_email_service
-from src.services.users import get_service as get_user_service
+from src.services.auth import AuthService
+from src.services.users import UserService
+from dataclasses import dataclass
 from src.infra.db import get_database
 from src.infra.users import UserRepoImpl
 from waitress import serve
@@ -15,20 +17,31 @@ from src.config import parse_config, Config
 logger = logging.getLogger(__name__)
 
 
+@dataclass
+class AppServices:
+    users: UserService
+    auth: AuthService
+    email: EmailService
+    notifications: NotificationService
+
+
+class MyFlask(Flask):
+    services: AppServices
+
+
 def create_app(name: str, config: Config) -> Flask:
     logging.basicConfig(level=config.logging.level, format=('%(asctime)s %(levelname)s [%(name)s] %(message)s'))
 
-    app = Flask(name)
+    app = MyFlask(name)
-
-    email_service = get_email_service(config.email)
-    notification_service = make_notification_service(email_service)
     database = get_database(config.database)
     user_repo = UserRepoImpl(database)
-    user_service = get_user_service(user_repo)
+    email_service = get_email_service(config.email)
-
+    app.services = AppServices(
-    app.extensions['email'] = email_service
+        users=UserService(user_repo),
-    app.extensions['notifications'] = notification_service
+        auth=AuthService(config.auth),
-    app.extensions['user_service'] = user_service
+        email=email_service,
+        notifications=get_notification_service(email_service),
+    )
 
     if config.host:
         app.config['SERVER_NAME'] = config.host

src/notifications/data.py

@@ -1,38 +0,0 @@
-from dataclasses import dataclass, field
-
-URL = str
-
-
-@dataclass
-class NotificationAction:
-    # A unique identifier for the action
-    action: str
-    # The title to show with the action
-    title: str
-    # Where to navigate to when clicked.
-    navigate: URL | None = None
-
-
-@dataclass
-class Notification:
-    """
-    Details to show in a notification
-
-    Much of the data here is intended to match the arguments to the [showNotification](https://developer.mozilla.org/en-US/docs/Web/API/ServiceWorkerRegistration/showNotification)
-    Web API, but this content is ideally displayable in other contexts (for example in Email or RSS)
-    """
-
-    # The main title content to show
-    title: str
-    # The content of the notification
-    content: str
-
-    # The UNIX timestamp in milliseconds since the epoch associated with this notification
-    timestamp: int
-    # A list of actions included with this notification
-    actions: list[NotificationAction] = field(default_factory=list)
-    # The location of an icon to show as the icon for this notification
-    icon: URL | None = None
-
-    # The location of an image associated with this notification
-    image: URL | None = None

src/services/auth.py

@@ -0,0 +1,45 @@
+from src.config.parse import assert_key_of_type, ParseError
+from typing import Any
+from dataclasses import dataclass, asdict
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+import jwt
+
+from src.services.users import User
+from src.config.auth import Auth as AuthConfig
+
+JWT = str
+
+
+@dataclass
+class Claims:
+    id: int
+
+    @classmethod
+    def from_user(cls, user: User) -> 'Claims':
+        return Claims(id=user.id)
+
+    @classmethod
+    def from_dict(cls, claims: dict[str, Any]) -> 'Claims':
+        assert_key_of_type(claims, 'id', int)
+        return Claims(id=claims['id'])
+
+
+class AuthService:
+    def __init__(self, config: AuthConfig):
+        self._private_key = Ed25519PrivateKey.from_private_bytes(config.ed25519_private_key.expose_secret())
+        self._public_key = self._private_key.public_key()
+
+    def mint_jwt(self, user: User) -> JWT:
+        claims = Claims.from_user(user)
+        return jwt.encode(asdict(claims), self._private_key, algorithm='EdDSA')
+
+    def validate_token(self, token: JWT) -> Claims | None:
+        try:
+            claims = jwt.decode(token, key=self._public_key, algorithms=['EdDSA'])
+        except jwt.InvalidTokenError:
+            return None
+
+        try:
+            return Claims.from_dict(claims)
+        except ParseError:
+            return None

src/services/email.py

@@ -42,5 +42,5 @@ class BirdEmailServiceImpl(EmailService):
         return response.ok
 
 
-def get_service(config: EmailConfig) -> EmailService:
+def get_email_service(config: EmailConfig) -> EmailService:
     return BirdEmailServiceImpl(config)

src/services/notifications.py

@@ -2,11 +2,45 @@ import abc
 from typing import Any
 
 from src.services.email import EmailService
-from src.config import Config
 
-from .data import Notification
+from dataclasses import dataclass, field
 
-# MUSTFIX: Needs real users!
+URL = str
+
+
+@dataclass
+class NotificationAction:
+    # A unique identifier for the action
+    action: str
+    # The title to show with the action
+    title: str
+    # Where to navigate to when clicked.
+    navigate: URL | None = None
+
+
+@dataclass
+class Notification:
+    """
+    Details to show in a notification
+
+    Much of the data here is intended to match the arguments to the [showNotification](https://developer.mozilla.org/en-US/docs/Web/API/ServiceWorkerRegistration/showNotification)
+    Web API, but this content is ideally displayable in other contexts (for example in Email or RSS)
+    """
+
+    # The main title content to show
+    title: str
+    # The content of the notification
+    content: str
+
+    # The UNIX timestamp in milliseconds since the epoch associated with this notification
+    timestamp: int
+    # A list of actions included with this notification
+    actions: list[NotificationAction] = field(default_factory=list)
+    # The location of an icon to show as the icon for this notification
+    icon: URL | None = None
+
+    # The location of an image associated with this notification
+    image: URL | None = None
 
 
 class NotificationSender(abc.ABC):
@@ -48,7 +82,7 @@ class EmailNotifier(NotificationSender):
         pass
 
 
-def make_notification_service(email_service: EmailService):
+def get_notification_service(email_service: EmailService):
     service = NotificationService()
     service.register_sender(EmailNotifier(email_service))
 

src/services/users.py

@@ -96,7 +96,3 @@ class UserService:
         user = self._repo.get_user_by_id(user_id)
         if user:
             return user.to_profile()
-
-
-def get_service(repo: UserRepo) -> UserService:
-    return UserService(repo)