Coverage for src/lexigram/admin/controllers/auth/core.py: 48%

1from __future__ import annotations 

2 

3import re 

4import secrets 

5 

6from starlette.requests import Request 

7 

8from lexigram.admin.auth.protocols import ( 

9 AdminAuthServiceProtocol, 

10 AdminCsrfServiceProtocol, 

11 AdminEmailOtpServiceProtocol, 

12 AdminEmailVerificationServiceProtocol, 

13 AdminMfaServiceProtocol, 

14 AdminPasswordResetServiceProtocol, 

15) 

16from lexigram.admin.auth.store import AdminUserStoreProtocol 

17from lexigram.admin.controllers.base import AdminController 

18from lexigram.admin.engine.renderer import AdminRenderer 

19from lexigram.admin.observability.admin_metrics import AdminMetrics 

20from lexigram.contracts.core import TaskManagerProtocol 

21from lexigram.logging import get_logger 

22 

23_CACHE_CONTROL_NO_STORE = {"Cache-Control": "no-store"} 

24 

25_DEFAULT_NEXT = "/admin/" 

26 

27_LEX_ERR_RE = re.compile(r"^\[LEX_ERR_[A-Z0-9_]+\]\s+") 

28 

29_SEE_DOCS_RE = re.compile(r"\n\s*→\s*See:.*$") 

30 

31 

32def _humanize_error(message: str) -> str: 

33 """Strip framework error prefixes from a message for user display. 

34 

35 Args: 

36 message: Raw error message, possibly including a ``[LEX_ERR_*]`` 

37 prefix and a trailing ``→ See:`` documentation line. 

38 

39 Returns: 

40 The message with the prefix and docs line removed, stripped. 

41 """ 

42 if not message: 

43 return "" 

44 return _SEE_DOCS_RE.sub("", _LEX_ERR_RE.sub("", message)).strip() 

45 

46 

47logger = get_logger(__name__) 

48 

49 

50class AuthCoreMixin(AdminController): 

51 """AuthController core: construction, CSRF, URL safety helpers.""" 

52 

53 def __init__( 

54 self, 

55 auth_service: AdminAuthServiceProtocol, 

56 csrf_service: AdminCsrfServiceProtocol, 

57 renderer: AdminRenderer, 

58 task_manager: TaskManagerProtocol | None = None, 

59 metrics: AdminMetrics | None = None, 

60 password_reset_service: AdminPasswordResetServiceProtocol | None = None, 

61 mfa_service: AdminMfaServiceProtocol | None = None, 

62 email_verification_service: AdminEmailVerificationServiceProtocol | None = None, 

63 email_otp_service: AdminEmailOtpServiceProtocol | None = None, 

64 ) -> None: 

65 """Initialise auth controller. 

66 

67 Args: 

68 auth_service: Orchestrates credential verification, rate limiting, 

69 lockout checks, session issuance, and audit logging. 

70 csrf_service: Generates and validates CSRF tokens. 

71 renderer: AdminRenderer for page rendering. 

72 task_manager: Optional task manager; injected by the container in 

73 production, omitted in tests. 

74 metrics: Optional admin metrics collector. 

75 password_reset_service: Optional password reset orchestrator; 

76 injected by the container when registered, ``None`` otherwise. 

77 mfa_service: Optional TOTP 2FA orchestrator; injected by the 

78 container when registered, ``None`` otherwise. 

79 email_verification_service: Optional email verification 

80 orchestrator; ``None`` disables the verification gate. 

81 email_otp_service: Optional email OTP factor orchestrator; 

82 ``None`` disables email-code challenges. 

83 """ 

84 super().__init__(renderer, task_manager) 

85 self._auth_service = auth_service 

86 self._csrf_service = csrf_service 

87 self._metrics = metrics or AdminMetrics(None) 

88 self._password_reset_service = password_reset_service 

89 self._mfa_service = mfa_service 

90 self._email_verification_service = email_verification_service 

91 self._email_otp_service = email_otp_service 

92 # Self-service registration — wired by the bundle provider from 

93 # ``AdminAuthConfig.registration`` (off unless explicitly enabled). 

94 self._user_store: AdminUserStoreProtocol | None = None 

95 self._registration_enabled = False 

96 self._registration_default_role = "admin" 

97 self._registration_domains: list[str] = [] 

98 

99 def _fresh_csrf(self, request: Request) -> str: 

100 """Generate a fresh CSRF token bound to a new session id.""" 

101 csrf_session_id = secrets.token_urlsafe(16) 

102 request.session["csrf_session_id"] = csrf_session_id 

103 return self._csrf_service.generate_token(csrf_session_id) 

104 

105 @staticmethod 

106 def _safe_next_url(candidate: str) -> str: 

107 """Return ``candidate`` if it is a safe same-origin relative path, else the default. 

108 

109 Rejects absolute URLs (any scheme), scheme-relative URLs (leading 

110 ``//``, interpreted by browsers as ``{current-scheme}://``), and 

111 anything not starting with a single ``/``. This is the sole point 

112 every post-login/post-MFA/post-verification ``next`` value must pass 

113 through before being stored in session state or used in a redirect. 

114 

115 Args: 

116 candidate: Raw ``next`` value from user input. 

117 

118 Returns: 

119 ``candidate`` when it starts with a single ``/`` (and not 

120 ``//`` or ``/\\``), otherwise ``_DEFAULT_NEXT``. 

121 """ 

122 if ( 

123 candidate 

124 and candidate.startswith("/") 

125 and not candidate.startswith("//") 

126 and not candidate.startswith("/\\") 

127 ): 

128 return candidate 

129 return _DEFAULT_NEXT 

130 

131 def _get_client_ip(self, request: Request) -> str: 

132 """Extract the real client IP from the request. 

133 

134 Prefers the first value of the ``X-Forwarded-For`` header when present 

135 (set by reverse proxies), falling back to the direct TCP peer address. 

136 

137 Args: 

138 request: Incoming HTTP request. 

139 

140 Returns: 

141 IP address string, or ``"unknown"`` when unavailable. 

142 """ 

143 forwarded = request.headers.get("x-forwarded-for", "") 

144 if forwarded: 

145 return forwarded.split(",")[0].strip() 

146 return request.client.host if request.client else "unknown"