Coverage for src / lexigram / admin / controllers / auth.py: 34%
79 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-13 22:14 +0800
1"""Built-in authentication controller for Lexigram Admin.
3Provides login/logout endpoints with standalone UI (no admin shell).
4"""
6from __future__ import annotations
8import secrets
9from urllib.parse import quote_plus
11from starlette.requests import Request
12from starlette.responses import HTMLResponse, RedirectResponse
14from lexigram.admin.auth.protocols import (
15 AdminAuthServiceProtocol,
16 AdminCsrfServiceProtocol,
17)
18from lexigram.admin.controllers.base import AdminController
19from lexigram.admin.engine.renderer import AdminRenderer
20from lexigram.admin.lib.template import render_login_page
21from lexigram.admin.observability.admin_metrics import AdminMetrics
22from lexigram.contracts.core import TaskManagerProtocol
23from lexigram.contracts.web import get, post
24from lexigram.di.decorators import inject
25from lexigram.logging import get_logger
27logger = get_logger(__name__)
30@inject
31class AuthController(AdminController):
32 """Built-in authentication controller.
34 Provides:
35 - GET /admin/login — Standalone login page with CSRF token
36 - POST /admin/login — Process login credentials through the full auth pipeline
37 - GET /admin/logout — Invalidate session and redirect to login
38 """
40 prefix = ""
42 def __init__(
43 self,
44 auth_service: AdminAuthServiceProtocol,
45 csrf_service: AdminCsrfServiceProtocol,
46 renderer: AdminRenderer,
47 task_manager: TaskManagerProtocol | None = None,
48 metrics: AdminMetrics | None = None,
49 ) -> None:
50 """Initialise auth controller.
52 Args:
53 auth_service: Orchestrates credential verification, rate limiting,
54 lockout checks, session issuance, and audit logging.
55 csrf_service: Generates and validates CSRF tokens.
56 renderer: AdminRenderer for page rendering.
57 task_manager: Optional task manager; injected by the container in
58 production, omitted in tests.
59 metrics: Optional admin metrics collector.
60 """
61 super().__init__(renderer, task_manager)
62 self._auth_service = auth_service
63 self._csrf_service = csrf_service
64 self._metrics = metrics or AdminMetrics(None)
66 # ------------------------------------------------------------------
67 # GET /login
68 # ------------------------------------------------------------------
70 @get("/login")
71 async def login_form(self, request: Request) -> HTMLResponse | RedirectResponse:
72 """Display the standalone login form.
74 Redirects authenticated users to *next_url*. For unauthenticated
75 visitors a fresh CSRF token is embedded in the rendered form.
77 Args:
78 request: Incoming HTTP request.
80 Returns:
81 HTMLResponse with the rendered login page, or a RedirectResponse
82 when the user is already authenticated.
83 """
84 next_url = request.query_params.get("next", "/admin/")
86 user = getattr(request.state, "user", None)
87 if user and user.user_id != "guest":
88 return RedirectResponse(url=next_url, status_code=302)
90 error = request.query_params.get("error", "")
92 csrf_session_id = secrets.token_urlsafe(16)
93 request.session["csrf_session_id"] = csrf_session_id
94 csrf_token = self._csrf_service.generate_token(csrf_session_id)
96 html = render_login_page(next_url=next_url, error=error, csrf_token=csrf_token)
97 return HTMLResponse(content=html)
99 # ------------------------------------------------------------------
100 # POST /login
101 # ------------------------------------------------------------------
103 @post("/login")
104 async def login_submit(self, request: Request) -> RedirectResponse:
105 """Process login form submission through the full security pipeline.
107 Validates the CSRF token, resolves the client IP, and delegates
108 credential verification to ``AdminAuthServiceProtocol.authenticate``.
109 On success the session is populated and the user is redirected to
110 *next_url*. On failure the login page is re-shown with a descriptive
111 error message.
113 Args:
114 request: Incoming HTTP request carrying form data.
116 Returns:
117 RedirectResponse to *next_url* on success, or back to the login
118 page with an error query parameter on failure.
119 """
120 form_data = await request.form()
121 email = str(form_data.get("email", ""))
122 password = str(form_data.get("password", ""))
123 next_url = str(form_data.get("next", "/admin/"))
124 csrf_token = str(form_data.get("csrf_token", ""))
126 # ── CSRF validation ────────────────────────────────────────────
127 csrf_session_id = request.session.get("csrf_session_id", "")
128 if not csrf_session_id or not self._csrf_service.validate_token(
129 csrf_session_id, csrf_token
130 ):
131 logger.warning(
132 "auth.csrf_validation_failed", ip=self._get_client_ip(request)
133 )
134 return RedirectResponse(
135 url=f"/admin/login?error={quote_plus('Invalid or expired security token. Please try again.')}&next={quote_plus(next_url)}",
136 status_code=302,
137 )
139 # ── Basic input guard ──────────────────────────────────────────
140 if not email or not password:
141 return RedirectResponse(
142 url=f"/admin/login?error={quote_plus('Email and password are required.')}&next={quote_plus(next_url)}",
143 status_code=302,
144 )
146 # ── Resolve client context ─────────────────────────────────────
147 ip = self._get_client_ip(request)
148 user_agent = request.headers.get("user-agent", "")
150 logger.info("auth.login_attempt", email=email, ip=ip)
152 result = await self._auth_service.authenticate(email, password, ip, user_agent)
154 if result.is_ok():
155 auth_result = result.unwrap()
156 request.session["admin_user_id"] = auth_result.user_id
157 request.session["admin_user_email"] = auth_result.email
158 if hasattr(auth_result, "session_id"):
159 request.session["session_id"] = auth_result.session_id
160 self._metrics.record_login(status="success")
161 logger.info(
162 "auth.login_success",
163 user_id=auth_result.user_id,
164 email=auth_result.email,
165 redirect=next_url,
166 )
167 return RedirectResponse(url=next_url, status_code=302)
169 error_msg = str(result.unwrap_err())
170 self._metrics.record_login(status="failure")
171 logger.warning("auth.login_failed", email=email, ip=ip, reason=error_msg)
172 return RedirectResponse(
173 url=f"/admin/login?error={quote_plus(error_msg)}&next={quote_plus(next_url)}",
174 status_code=302,
175 )
177 # ------------------------------------------------------------------
178 # GET /logout
179 # ------------------------------------------------------------------
181 @get("/logout")
182 async def logout(self, request: Request) -> RedirectResponse:
183 """Invalidate the current session and redirect to the login page.
185 Calls ``AdminAuthServiceProtocol.invalidate_session`` before clearing
186 the Starlette session cookie so that any server-side session record is
187 also revoked.
189 Args:
190 request: Incoming HTTP request.
192 Returns:
193 RedirectResponse to /admin/login.
194 """
195 session_id = request.session.get("session_id")
196 if session_id:
197 await self._auth_service.invalidate_session(session_id)
198 logger.info("auth.logout", session_id=session_id)
200 request.session.clear()
201 return RedirectResponse(url="/admin/login", status_code=302)
203 # ------------------------------------------------------------------
204 # Helpers
205 # ------------------------------------------------------------------
207 def _get_client_ip(self, request: Request) -> str:
208 """Extract the real client IP from the request.
210 Prefers the first value of the ``X-Forwarded-For`` header when present
211 (set by reverse proxies), falling back to the direct TCP peer address.
213 Args:
214 request: Incoming HTTP request.
216 Returns:
217 IP address string, or ``"unknown"`` when unavailable.
218 """
219 forwarded = request.headers.get("x-forwarded-for", "")
220 if forwarded:
221 return forwarded.split(",")[0].strip()
222 return request.client.host if request.client else "unknown"
225__all__ = ["AuthController"]