Add an auth service for minting and decoding JWTs

This commit is contained in:
Campbell Alden 2026-08-02 22:50:05 +09:00
parent 328f430309
commit 387e2fa490
6 changed files with 77 additions and 4 deletions

View file

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

View file

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

View file

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

View file

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

16
src/config/auth.py Normal file
View file

@ -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))

45
src/services/auth.py Normal file
View file

@ -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