"""Authentication: Google OAuth (Authlib) + opt-in dev-login. Role resolution on login: - email == ADMIN_EMAIL -> admin (the Lead), auto-created on first login - email in subordinate records -> subordinate (must be active) - otherwise -> access denied """ from functools import wraps from authlib.integrations.flask_client import OAuth from flask import ( Blueprint, abort, current_app, flash, redirect, render_template, request, url_for, ) from flask_login import ( LoginManager, UserMixin, current_user, login_required, login_user, logout_user, ) from . import storage auth_bp = Blueprint("auth", __name__) login_manager = LoginManager() oauth = OAuth() google = None class User(UserMixin): """Flask-Login wrapper around a stored user record.""" def __init__(self, record): self.record = record def get_id(self): return str(self.record["id"]) @property def id(self): return self.record["id"] @property def name(self): return self.record["name"] @property def email(self): return self.record["email"] @property def role(self): return self.record["role"] @property def is_admin(self): return self.record["role"] == "admin" @login_manager.user_loader def load_user(user_id): try: record = storage.get_user(int(user_id)) except (TypeError, ValueError): return None if record is None or not record.get("active", True): return None return User(record) def init_auth(app): login_manager.init_app(app) login_manager.login_view = "auth.login" login_manager.login_message = "Please sign in to continue." oauth.init_app(app) global google if app.config.get("GOOGLE_CLIENT_ID") and app.config.get("GOOGLE_CLIENT_SECRET"): google = oauth.register( name="google", client_id=app.config["GOOGLE_CLIENT_ID"], client_secret=app.config["GOOGLE_CLIENT_SECRET"], server_metadata_url="https://accounts.google.com/.well-known/openid-configuration", client_kwargs={"scope": "openid email profile"}, ) app.logger.info( "Google OAuth configured (redirect_uri=%s/auth/callback).", app.config.get("BASE_URL"), ) else: missing = [ k for k in ("GOOGLE_CLIENT_ID", "GOOGLE_CLIENT_SECRET") if not app.config.get(k) ] app.logger.warning( "Google OAuth DISABLED: missing %s in the environment. " "Sign-in will show 'not configured'.", ", ".join(missing), ) def admin_required(view): @wraps(view) @login_required def wrapped(*args, **kwargs): if not current_user.is_admin: abort(403) return view(*args, **kwargs) return wrapped def _persist_login(user): """Log the user in with a long-lived remember cookie so the session survives browser/PWA restarts and cookie eviction — they stay signed in until they explicitly log out.""" from flask import session session.permanent = True login_user(user, remember=True) def _resolve_and_login(email, name): """Given a verified email, log the user in per the role rules.""" email = (email or "").strip().lower() admin_emails = current_app.config.get("ADMIN_EMAILS", []) if email and email in admin_emails: record = storage.get_user_by_email(email) if record is None: record = storage.create_user(email, name or "Lead", "admin") elif not record.get("active", True): storage.update_user(record["id"], active=True) record = storage.get_user(record["id"]) _persist_login(User(record)) return True record = storage.get_user_by_email(email) if record is not None and record["role"] == "subordinate" and record.get("active", True): # Keep the display name fresh from the identity provider. if name and name != record["name"]: storage.update_user(record["id"], name=name) record = storage.get_user(record["id"]) _persist_login(User(record)) return True return False # -------------------------------------------------------------------------- # Routes # -------------------------------------------------------------------------- @auth_bp.route("/login") def login(): if current_user.is_authenticated: return redirect(url_for("tasks.home")) return render_template( "login.html", google_enabled=google is not None, dev_login=current_app.config.get("ALLOW_DEV_LOGIN", False), subordinates=storage.get_subordinates() if current_app.config.get("ALLOW_DEV_LOGIN") else [], ) @auth_bp.route("/login/google") def login_google(): if google is None: flash("Google sign-in is not configured yet.", "error") return redirect(url_for("auth.login")) redirect_uri = current_app.config["BASE_URL"] + url_for("auth.callback") return google.authorize_redirect(redirect_uri) @auth_bp.route("/auth/callback") def callback(): if google is None: abort(404) token = google.authorize_access_token() userinfo = token.get("userinfo") or google.userinfo() email = userinfo.get("email") name = userinfo.get("name") or userinfo.get("given_name") or "" if not userinfo.get("email_verified", True): return render_template("denied.html", email=email), 403 if _resolve_and_login(email, name): return redirect(url_for("tasks.home")) return render_template("denied.html", email=email), 403 @auth_bp.route("/login/dev", methods=["POST"]) def login_dev(): """Opt-in local login without Google. Only active when ALLOW_DEV_LOGIN=true.""" if not current_app.config.get("ALLOW_DEV_LOGIN"): abort(404) role = request.form.get("role") if role == "admin": admin_emails = current_app.config.get("ADMIN_EMAILS", []) email = admin_emails[0] if admin_emails else "admin@example.com" if not _resolve_and_login(email, "Lead"): flash("Set ADMIN_EMAIL in your .env to use admin dev-login.", "error") return redirect(url_for("auth.login")) return redirect(url_for("tasks.home")) try: user_id = int(request.form.get("user_id", "")) except ValueError: abort(400) record = storage.get_user(user_id) if record is None or not record.get("active", True): abort(404) _persist_login(User(record)) return redirect(url_for("tasks.home")) @auth_bp.route("/logout") @login_required def logout(): logout_user() return redirect(url_for("auth.login"))