Spaces:
Running
Running
| """ | |
| api/auth.py -- sign-in endpoints (auth-plan.md). | |
| GET /auth/config public: what the login page needs (client id, domain, dev flag) | |
| POST /auth/google Google ID token -> our session token (+ the user) | |
| POST /auth/signup name + approved email + password -> creates the account and signs in | |
| POST /auth/login approved email + password -> session token | |
| POST /auth/dev-login LOCAL TESTING ONLY (AUTH_DEV_LOGIN=true): email -> session token | |
| GET /auth/me who am I (401 when not signed in) | |
| GET /users active team members (for @mentions and the team tables) | |
| """ | |
| from __future__ import annotations | |
| from fastapi import APIRouter, Depends, HTTPException, status | |
| from sqlalchemy.orm import Session | |
| from app.api.deps import get_current_user, get_db, require_access | |
| from app.config import settings | |
| from app.crud import users as crud_users | |
| from app.models import User | |
| from app.schemas.auth import ( | |
| AuthConfig, | |
| DevLoginInput, | |
| GoogleLoginInput, | |
| LoginOut, | |
| PasswordLoginInput, | |
| SignupInput, | |
| UserOut, | |
| ) | |
| from app.services import auth as auth_service | |
| router = APIRouter(tags=["auth"]) | |
| def auth_config(): | |
| return AuthConfig( | |
| google_client_id=(settings.google_client_id or "").strip(), | |
| allowed_domains=settings.allowed_domain_list, | |
| auth_required=settings.auth_required, | |
| dev_login=settings.auth_dev_login, | |
| signup_code_required=bool((settings.signup_code or "").strip()), | |
| min_password_length=auth_service.MIN_PASSWORD_LENGTH, | |
| ) | |
| def _issue(db: Session, email: str, name: str, picture: str) -> LoginOut: | |
| user = crud_users.sign_in(db, email, name, picture) | |
| if user is None: | |
| raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="This account has been deactivated.") | |
| try: | |
| token = auth_service.create_token(user.id, user.email, user.name) | |
| except auth_service.AuthError as exc: | |
| raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(exc)) from exc | |
| return LoginOut(token=token, user=UserOut.from_model(user)) | |
| def google_login(payload: GoogleLoginInput, db: Session = Depends(get_db)): | |
| try: | |
| who = auth_service.verify_google_credential(payload.credential) | |
| except auth_service.AuthError as exc: | |
| raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=str(exc)) from exc | |
| return _issue(db, who["email"], who["name"], who["picture"]) | |
| _NOT_APPROVED = "This email is not on the approved list. Ask the admin to add it." | |
| def signup(payload: SignupInput, db: Session = Depends(get_db)): | |
| """Create an account with a name, an approved email and a password, and sign in.""" | |
| email = payload.email.strip().lower() | |
| if not auth_service.email_allowed(email): | |
| raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=_NOT_APPROVED) | |
| if not auth_service.signup_code_ok(payload.code): | |
| raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="The invite code is wrong.") | |
| if crud_users.get_user_by_email(db, email) is not None: | |
| raise HTTPException( | |
| status_code=status.HTTP_409_CONFLICT, | |
| detail="An account with this email already exists. Sign in instead.", | |
| ) | |
| user = crud_users.sign_in(db, email, payload.name, "") | |
| crud_users.set_password(db, user, auth_service.hash_password(payload.password)) | |
| return _issue(db, email, "", "") | |
| def password_login(payload: PasswordLoginInput, db: Session = Depends(get_db)): | |
| email = payload.email.strip().lower() | |
| if auth_service.login_throttled(email): | |
| raise HTTPException( | |
| status_code=status.HTTP_429_TOO_MANY_REQUESTS, | |
| detail="Too many wrong attempts. Wait 10 minutes and try again.", | |
| ) | |
| user = crud_users.get_user_by_email(db, email) | |
| # Same work and same message whether the email exists or not, so the form cannot be used to | |
| # find out who has an account. | |
| ok = auth_service.verify_password(payload.password, (user.password_hash if user else None) or auth_service.dummy_hash()) | |
| if not ok: | |
| auth_service.note_login_failure(email) | |
| raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Wrong email or password.") | |
| # Taking an email off the approved list locks that person out even though the account exists. | |
| if not auth_service.email_allowed(email): | |
| raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=_NOT_APPROVED) | |
| auth_service.clear_login_failures(email) | |
| return _issue(db, email, "", "") | |
| def dev_login(payload: DevLoginInput, db: Session = Depends(get_db)): | |
| # Hidden (404) unless explicitly enabled, so a production service never advertises it. | |
| if not settings.auth_dev_login: | |
| raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found") | |
| email = payload.email.strip().lower() | |
| if not auth_service.email_allowed(email): | |
| raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=_NOT_APPROVED) | |
| return _issue(db, email, email.split("@")[0].replace(".", " ").title(), "") | |
| def me(user: User | None = Depends(get_current_user)): | |
| if user is None: | |
| raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not signed in.") | |
| return UserOut.from_model(user) | |
| def list_users(db: Session = Depends(get_db)): | |
| return [UserOut.from_model(u) for u in crud_users.list_users(db, active_only=True)] | |