139 lines
5.3 KiB
Python
139 lines
5.3 KiB
Python
"""Authentication service.
|
|
|
|
Handles registration, login, token generation, and current-user lookup.
|
|
Passwords hashed with argon2 (via argon2-cffi). Tokens are short-lived JWTs
|
|
(access) + httpOnly refresh tokens. Refresh tokens are NOT persisted in MVP
|
|
(stateless — revoke by letting them expire). Add a refresh_token table when
|
|
true revocation is needed.
|
|
|
|
TODO (post-MVP): password reset + email verification flows are not implemented.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from uuid import UUID, uuid4
|
|
|
|
import jwt
|
|
from argon2 import PasswordHasher
|
|
from argon2.exceptions import VerifyMismatchError
|
|
|
|
from app.config import settings
|
|
from app.entities.personal_data import PersonalData
|
|
from app.entities.user import User
|
|
from app.entities.workspace import Workspace
|
|
from app.repositories.personal_data_repository import PersonalDataRepository
|
|
from app.repositories.user_repository import UserRepository
|
|
from app.repositories.workspace_repository import WorkspaceRepository
|
|
from app.utils.errors import AuthError, ConflictError, UnauthorizedError
|
|
|
|
_ph = PasswordHasher()
|
|
|
|
|
|
class AuthService:
|
|
def __init__(
|
|
self,
|
|
user_repo: UserRepository,
|
|
personal_data_repo: PersonalDataRepository,
|
|
workspace_repo: WorkspaceRepository,
|
|
) -> None:
|
|
self._users = user_repo
|
|
self._personal_data = personal_data_repo
|
|
self._workspaces = workspace_repo
|
|
|
|
async def register(
|
|
self, email: str, password: str, display_name: str
|
|
) -> tuple[User, PersonalData]:
|
|
"""Create User + personal workspace + PersonalData. Raises ConflictError if email taken."""
|
|
existing = await self._users.get_by_email(email)
|
|
if existing is not None:
|
|
raise ConflictError(f"Email already registered: {email}")
|
|
|
|
# Create a personal workspace for the registering user.
|
|
workspace = Workspace(
|
|
id=uuid4(),
|
|
name=f"{display_name}s planering",
|
|
created_at=datetime.now(),
|
|
)
|
|
workspace = await self._workspaces.create(workspace)
|
|
|
|
pd = PersonalData(
|
|
uuid=uuid4(),
|
|
workspace_id=workspace.id,
|
|
name=display_name,
|
|
created_at=datetime.now(),
|
|
)
|
|
pd = await self._personal_data.create(pd)
|
|
|
|
user = User(
|
|
id=uuid4(),
|
|
personal_data_uuid=pd.uuid,
|
|
email=email,
|
|
hashed_password=_ph.hash(password),
|
|
created_at=datetime.now(),
|
|
)
|
|
user = await self._users.create(user)
|
|
return user, pd
|
|
|
|
async def login(self, email: str, password: str) -> User:
|
|
"""Verify credentials. Raises AuthError on wrong email or password."""
|
|
user = await self._users.get_by_email(email)
|
|
if user is None:
|
|
raise AuthError("Invalid email or password")
|
|
try:
|
|
_ph.verify(user.hashed_password, password)
|
|
except VerifyMismatchError as exc:
|
|
raise AuthError("Invalid email or password") from exc
|
|
if _ph.check_needs_rehash(user.hashed_password):
|
|
user.hashed_password = _ph.hash(password)
|
|
await self._users.update(user.id, user)
|
|
return user
|
|
|
|
def create_access_token(self, user_id: UUID) -> str:
|
|
expire = datetime.now(timezone.utc) + timedelta(
|
|
minutes=settings.jwt_access_token_expire_minutes
|
|
)
|
|
payload = {"sub": str(user_id), "exp": expire, "type": "access"}
|
|
return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)
|
|
|
|
def create_refresh_token(self, user_id: UUID) -> str:
|
|
expire = datetime.now(timezone.utc) + timedelta(days=settings.jwt_refresh_token_expire_days)
|
|
payload = {"sub": str(user_id), "exp": expire, "type": "refresh"}
|
|
return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)
|
|
|
|
async def get_user_from_token(self, token: str) -> User:
|
|
"""Decode a JWT access token and return the user. Raises UnauthorizedError on failure."""
|
|
try:
|
|
payload = jwt.decode(token, settings.jwt_secret, algorithms=[settings.jwt_algorithm])
|
|
except jwt.ExpiredSignatureError as exc:
|
|
raise UnauthorizedError("Token expired") from exc
|
|
except jwt.InvalidTokenError as exc:
|
|
raise UnauthorizedError("Invalid token") from exc
|
|
|
|
if payload.get("type") != "access":
|
|
raise UnauthorizedError("Expected access token")
|
|
|
|
user_id = UUID(payload["sub"])
|
|
user = await self._users.get_by_id(user_id)
|
|
if user is None:
|
|
raise UnauthorizedError("User not found")
|
|
return user
|
|
|
|
async def refresh_access_token(self, refresh_token: str) -> str:
|
|
"""Issue a new access token from a valid refresh token."""
|
|
try:
|
|
payload = jwt.decode(
|
|
refresh_token, settings.jwt_secret, algorithms=[settings.jwt_algorithm]
|
|
)
|
|
except jwt.InvalidTokenError as exc:
|
|
raise UnauthorizedError("Invalid refresh token") from exc
|
|
|
|
if payload.get("type") != "refresh":
|
|
raise UnauthorizedError("Expected refresh token")
|
|
|
|
user_id = UUID(payload["sub"])
|
|
user = await self._users.get_by_id(user_id)
|
|
if user is None:
|
|
raise UnauthorizedError("User not found")
|
|
return self.create_access_token(user.id)
|