Coverage for src / lexigram / admin / controllers / setup.py: 28%

1"""First-run setup controller for Lexigram Admin. 

2 

3Provides the initial account creation wizard shown when no admin users exist. 

4The SetupMiddleware redirects all admin requests here until at least one 

5admin account has been created. 

6""" 

7 

8from __future__ import annotations 

9 

10import hashlib 

11import os 

12 

13from starlette.requests import Request 

14from starlette.responses import HTMLResponse, RedirectResponse 

15 

16from lexigram.admin.auth.protocols import ( 

17 AdminAuditLogServiceProtocol, 

18 AdminPasswordPolicyServiceProtocol, 

19) 

20from lexigram.admin.auth.store import AdminUserStoreProtocol 

21from lexigram.admin.auth.types import AdminSecurityEventType 

22from lexigram.admin.controllers.base import AdminController 

23from lexigram.admin.engine.renderer import AdminRenderer 

24from lexigram.admin.lib.template import render_setup_page 

25from lexigram.contracts.core import TaskManagerProtocol 

26from lexigram.contracts.web import get, post 

27from lexigram.di.decorators import inject 

28from lexigram.logging import get_logger 

29 

30logger = get_logger(__name__) 

31 

32 

33@inject 

34class SetupController(AdminController): 

35 """First-run setup wizard controller. 

36 

37 Provides: 

38 - GET /setup — Display account creation form 

39 - POST /setup — Create the first admin account 

40 """ 

41 

42 prefix = "" 

43 

44 def __init__( 

45 self, 

46 user_store: AdminUserStoreProtocol, 

47 password_policy_service: AdminPasswordPolicyServiceProtocol, 

48 audit_service: AdminAuditLogServiceProtocol, 

49 renderer: AdminRenderer, 

50 task_manager: TaskManagerProtocol | None = None, 

51 ) -> None: 

52 """Initialise setup controller. 

53 

54 Args: 

55 user_store: Store used to check and create admin accounts. 

56 password_policy_service: Validates passwords against all configured 

57 policy rules; returns every violation, not just the first. 

58 audit_service: Records security events; guaranteed never to raise. 

59 renderer: AdminRenderer required by AdminController base. 

60 task_manager: Optional; injected by container in production. 

61 """ 

62 super().__init__(renderer, task_manager) 

63 self._user_store = user_store 

64 self._password_policy_service = password_policy_service 

65 self._audit_service = audit_service 

66 

67 # ------------------------------------------------------------------ 

68 # GET /setup 

69 # ------------------------------------------------------------------ 

70 

71 @get("/setup") 

72 async def setup_form(self, request: Request) -> HTMLResponse | RedirectResponse: 

73 """Display the first-run setup form. 

74 

75 If at least one admin account already exists, a locked message is shown 

76 so the user knows to log in with their existing credentials. 

77 

78 Args: 

79 request: Incoming HTTP request. 

80 

81 Returns: 

82 HTMLResponse with the rendered setup page. 

83 """ 

84 try: 

85 count = await self._user_store.get_admin_count() 

86 except (RuntimeError, ValueError, OSError) as e: 

87 logger.warning("setup.count_failed error=%s", e) 

88 html = render_setup_page( 

89 error="Unable to verify setup status. Database may be unavailable." 

90 ) 

91 return HTMLResponse(content=html, status_code=503) 

92 if count > 0: 

93 html = render_setup_page( 

94 locked=True, 

95 error="Setup is already complete. Please log in with your existing account.", 

96 ) 

97 return HTMLResponse(content=html, status_code=200) 

98 

99 error = request.query_params.get("error", "") 

100 html = render_setup_page(error=error) 

101 return HTMLResponse(content=html) 

102 

103 # ------------------------------------------------------------------ 

104 # POST /setup 

105 # ------------------------------------------------------------------ 

106 

107 @post("/setup") 

108 async def setup_submit(self, request: Request) -> HTMLResponse | RedirectResponse: 

109 """Process the first-run setup form and create the initial admin account. 

110 

111 Validates the optional setup token, enforces the full password policy 

112 (all violations reported simultaneously), hashes the password with 

113 bcrypt, persists the user, audits the outcome, then redirects to the 

114 login page. 

115 

116 Args: 

117 request: Incoming HTTP request carrying form data. 

118 

119 Returns: 

120 RedirectResponse to ``/admin/login?next=/admin/`` on success, or 

121 an HTMLResponse re-rendering the setup form with error details on 

122 any validation or persistence failure. 

123 """ 

124 try: 

125 count = await self._user_store.get_admin_count() 

126 except (RuntimeError, ValueError, OSError) as e: 

127 logger.warning("setup.count_failed error=%s", e) 

128 html = render_setup_page( 

129 error="Unable to verify setup status. Database may be unavailable." 

130 ) 

131 return HTMLResponse(content=html, status_code=503) 

132 if count > 0: 

133 html = render_setup_page( 

134 locked=True, 

135 error="Setup is already complete. Please log in with your existing account.", 

136 ) 

137 return HTMLResponse(content=html, status_code=200) 

138 

139 form_data = await request.form() 

140 name = str(form_data.get("name", "")).strip() 

141 email = str(form_data.get("email", "")).strip() 

142 password = str(form_data.get("password", "")).strip() 

143 confirm = str(form_data.get("confirm_password", "")).strip() 

144 setup_token_input = str(form_data.get("setup_token", "")).strip() 

145 

146 ip = self._get_client_ip(request) 

147 user_agent = request.headers.get("user-agent", "") 

148 

149 # ── Optional setup-token guard ───────────────────────────────── 

150 required_token = os.environ.get("ADMIN_SETUP_TOKEN", "") 

151 if required_token and setup_token_input != required_token: 

152 logger.warning("setup.token_mismatch", ip=ip) 

153 await self._audit_service.log_event( 

154 event_type=AdminSecurityEventType.SETUP_BLOCKED, 

155 ip_address=ip, 

156 user_agent=user_agent, 

157 success=False, 

158 metadata={"reason": "invalid_setup_token"}, 

159 ) 

160 html = render_setup_page(error="Invalid setup token.") 

161 return HTMLResponse(content=html, status_code=403) 

162 

163 # ── Basic field presence ─────────────────────────────────────── 

164 if not name or not email or not password: 

165 html = render_setup_page(error="All fields are required.") 

166 return HTMLResponse(content=html, status_code=422) 

167 

168 if password != confirm: 

169 html = render_setup_page(error="Passwords do not match.") 

170 return HTMLResponse(content=html, status_code=422) 

171 

172 # ── Full password policy validation (all violations) ─────────── 

173 policy_result = self._password_policy_service.validate(password, email=email) 

174 if not policy_result.is_valid: 

175 violation_lines = "\n".join( 

176 f"• {v.message}" for v in policy_result.violations 

177 ) 

178 html = render_setup_page(error=violation_lines) 

179 return HTMLResponse(content=html, status_code=422) 

180 

181 # ── Hash and persist ─────────────────────────────────────────── 

182 hashed_password = _hash_password(password) 

183 

184 try: 

185 await self._user_store.create_user( 

186 name=name, 

187 email=email, 

188 hashed_password=hashed_password, 

189 roles=["superadmin"], 

190 ) 

191 except Exception as exc: 

192 # Treat any persistence failure (duplicate email, DB error, etc.) 

193 # as a non-fatal setup error that is shown back to the user. 

194 logger.error("setup.create_user_failed", email=email, error=str(exc)) 

195 html = render_setup_page(error=f"Failed to create account: {exc}") 

196 return HTMLResponse(content=html, status_code=422) 

197 

198 logger.info("setup.first_admin_created", email=email) 

199 

200 await self._audit_service.log_event( 

201 event_type=AdminSecurityEventType.SETUP_COMPLETED, 

202 ip_address=ip, 

203 user_agent=user_agent, 

204 success=True, 

205 metadata={"email": email}, 

206 ) 

207 

208 return RedirectResponse(url="/admin/login?next=/admin/", status_code=302) 

209 

210 # ------------------------------------------------------------------ 

211 # Helpers 

212 # ------------------------------------------------------------------ 

213 

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

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

216 

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

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

219 

220 Args: 

221 request: Incoming HTTP request. 

222 

223 Returns: 

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

225 """ 

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

227 if forwarded: 

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

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

230 

231 

232def _hash_password(plain: str) -> str: 

233 """Hash a plain-text password using bcrypt, falling back to SHA-256. 

234 

235 Bcrypt with 12 rounds is the default. SHA-256 is used only when the 

236 ``bcrypt`` package is not installed (e.g. lightweight test environments). 

237 

238 Args: 

239 plain: Plain-text password string. 

240 

241 Returns: 

242 Hashed password string suitable for storage. 

243 """ 

244 try: 

245 import bcrypt 

246 

247 hashed = bcrypt.hashpw(plain.encode("utf-8"), bcrypt.gensalt(rounds=12)) 

248 return hashed.decode("utf-8") 

249 except ImportError: 

250 return hashlib.sha256(plain.encode()).hexdigest() 

251 

252 

253__all__ = ["SetupController"]