| """ |
| Auth service wrapping Supabase Auth. |
| |
| Password handling: we never touch raw passwords beyond passing them straight |
| through to Supabase's signup/login calls over HTTPS - Supabase Auth hashes |
| and stores them (bcrypt) and issues session tokens. This app never stores a |
| password itself. |
| |
| In demo mode (no Supabase configured), a minimal local auth stand-in is used |
| so the app can boot and be exercised end-to-end without a real project. It |
| is intentionally simple and NOT secure - see the loud warnings below - and |
| must never be used in production. |
| """ |
| from __future__ import annotations |
|
|
| import hashlib |
| import re |
| import secrets |
| import threading |
| import uuid |
| from dataclasses import dataclass |
|
|
| from app.config import SUPABASE_CONFIGURED |
|
|
| MIN_PASSWORD_LENGTH = 10 |
| _PASSWORD_RULE = re.compile(r"^(?=.*[A-Za-z])(?=.*\d).{10,}$") |
|
|
|
|
| class AuthError(Exception): |
| pass |
|
|
|
|
| def validate_password_strength(password: str) -> None: |
| """Minimum password strength policy enforced on the signup form, |
| in addition to enabling Supabase's 'leaked password protection' setting |
| (a dashboard setting - see DEPLOY.md - which checks against known-breached |
| password lists server-side).""" |
| if not password or not _PASSWORD_RULE.match(password): |
| raise AuthError( |
| f"Password must be at least {MIN_PASSWORD_LENGTH} characters and contain " |
| f"both letters and numbers." |
| ) |
|
|
|
|
| @dataclass |
| class Session: |
| user_id: str |
| email: str |
| access_token: str |
|
|
|
|
| class AuthProvider: |
| def sign_up(self, email: str, password: str) -> Session: ... |
| def sign_in(self, email: str, password: str) -> Session: ... |
| def sign_out(self, access_token: str) -> None: ... |
| def get_user_from_token(self, access_token: str) -> tuple[str, str] | None: ... |
| def request_password_reset(self, email: str) -> None: ... |
| def delete_user(self, user_id: str) -> None: |
| """Permanently remove the auth user and invalidate all of their |
| sessions/tokens. Required for GDPR right-to-erasure: without this, |
| a still-valid token could keep authenticating after the account's |
| data rows were deleted, effectively resurrecting the account.""" |
| ... |
|
|
|
|
| class SupabaseAuthProvider(AuthProvider): |
| def __init__(self): |
| from supabase import create_client |
| import os |
|
|
| self.client = create_client(os.environ["SUPABASE_URL"], os.environ["SUPABASE_ANON_KEY"]) |
|
|
| def sign_up(self, email: str, password: str) -> Session: |
| validate_password_strength(password) |
| try: |
| res = self.client.auth.sign_up({"email": email, "password": password}) |
| except Exception as e: |
| raise AuthError(f"Sign up failed: {e}") from e |
| if not res.user or not res.session: |
| raise AuthError("Sign up requires email confirmation - check your inbox.") |
| return Session(user_id=res.user.id, email=email, access_token=res.session.access_token) |
|
|
| def sign_in(self, email: str, password: str) -> Session: |
| try: |
| res = self.client.auth.sign_in_with_password({"email": email, "password": password}) |
| except Exception as e: |
| raise AuthError("Invalid email or password.") from e |
| return Session(user_id=res.user.id, email=email, access_token=res.session.access_token) |
|
|
| def sign_out(self, access_token: str) -> None: |
| try: |
| self.client.auth.sign_out() |
| except Exception: |
| pass |
|
|
| def get_user_from_token(self, access_token: str) -> tuple[str, str] | None: |
| try: |
| res = self.client.auth.get_user(access_token) |
| if res and res.user: |
| return res.user.id, res.user.email |
| except Exception: |
| return None |
| return None |
|
|
| def request_password_reset(self, email: str) -> None: |
| try: |
| self.client.auth.reset_password_email(email) |
| except Exception as e: |
| raise AuthError(f"Password reset request failed: {e}") from e |
|
|
| def delete_user(self, user_id: str) -> None: |
| |
| |
| |
| |
| |
| |
| pass |
|
|
|
|
| class DemoAuthProvider(AuthProvider): |
| """NOT FOR PRODUCTION. In-memory auth used only when Supabase isn't |
| configured, so the app can be demoed offline. Passwords are hashed with |
| salted SHA-256 here ONLY as a minimal safeguard for the sandbox demo - |
| this is not a substitute for Supabase Auth / bcrypt and must never be |
| treated as production-grade. |
| """ |
|
|
| def __init__(self): |
| self._lock = threading.Lock() |
| self._users_by_email: dict[str, dict] = {} |
| self._tokens: dict[str, str] = {} |
|
|
| @staticmethod |
| def _hash(password: str, salt: str) -> str: |
| return hashlib.sha256((salt + password).encode()).hexdigest() |
|
|
| def sign_up(self, email: str, password: str) -> Session: |
| validate_password_strength(password) |
| with self._lock: |
| if email in self._users_by_email: |
| raise AuthError("An account with this email already exists.") |
| salt = secrets.token_hex(16) |
| user_id = str(uuid.uuid4()) |
| self._users_by_email[email] = { |
| "user_id": user_id, "email": email, "salt": salt, "hash": self._hash(password, salt), |
| } |
| return self._issue_session(user_id, email) |
|
|
| def sign_in(self, email: str, password: str) -> Session: |
| user = self._users_by_email.get(email) |
| if not user or self._hash(password, user["salt"]) != user["hash"]: |
| raise AuthError("Invalid email or password.") |
| return self._issue_session(user["user_id"], email) |
|
|
| def _issue_session(self, user_id: str, email: str) -> Session: |
| token = secrets.token_urlsafe(32) |
| self._tokens[token] = user_id |
| return Session(user_id=user_id, email=email, access_token=token) |
|
|
| def sign_out(self, access_token: str) -> None: |
| self._tokens.pop(access_token, None) |
|
|
| def get_user_from_token(self, access_token: str) -> tuple[str, str] | None: |
| user_id = self._tokens.get(access_token) |
| if not user_id: |
| return None |
| for u in self._users_by_email.values(): |
| if u["user_id"] == user_id: |
| return user_id, u["email"] |
| return None |
|
|
| def request_password_reset(self, email: str) -> None: |
| |
| |
| return None |
|
|
| def delete_user(self, user_id: str) -> None: |
| with self._lock: |
| email_to_remove = next( |
| (e for e, u in self._users_by_email.items() if u["user_id"] == user_id), None |
| ) |
| if email_to_remove: |
| self._users_by_email.pop(email_to_remove, None) |
| tokens_to_remove = [t for t, uid in self._tokens.items() if uid == user_id] |
| for t in tokens_to_remove: |
| self._tokens.pop(t, None) |
|
|
|
|
| _provider_instance: AuthProvider | None = None |
|
|
|
|
| def get_auth_provider() -> AuthProvider: |
| global _provider_instance |
| if _provider_instance is None: |
| _provider_instance = SupabaseAuthProvider() if SUPABASE_CONFIGURED else DemoAuthProvider() |
| return _provider_instance |
|
|