Coverage for core / src / sensorkit / webapi / security.py: 97%

1# SPDX-License-Identifier: Apache-2.0 

2import os 

3import pathlib 

4import secrets 

5import ssl 

6from abc import ABC, abstractmethod 

7from typing import Annotated, Any, Literal, override 

8 

9from fastapi.responses import JSONResponse 

10from loguru import logger 

11from pydantic import BaseModel, Field, model_validator 

12 

13BEARER_PREFIX = "Bearer " 

14READ_METHODS = frozenset({"GET", "HEAD", "OPTIONS"}) 

15LOOPBACK_HOSTS = frozenset({"127.0.0.1", "::1", "localhost"}) 

16DOC_PATHS = frozenset({"/openapi.json", "/docs", "/docs/oauth2-redirect", "/redoc"}) 

17SHORT_TOKEN_LENGTH = 16 

18 

19TLS_VERSIONS = { 

20 "1.2": ssl.TLSVersion.TLSv1_2, 

21 "1.3": ssl.TLSVersion.TLSv1_3, 

22} 

23 

24 

25class SecurityConfigError(Exception): 

26 """The security configuration cannot be applied on this host.""" 

27 

28 

29class TLSConfig(BaseModel): 

30 """TLS settings for the web API listener. 

31 

32 Key material stays on disk. This record is written to the key-value store, which 

33 every service on the bus can read, so it carries file paths and the name of the 

34 environment variable holding the key password rather than the secrets themselves. 

35 """ 

36 

37 certfile: pathlib.Path 

38 keyfile: pathlib.Path 

39 keyfile_password_env: str | None = None 

40 ca_certs: pathlib.Path | None = None 

41 require_client_cert: bool = False 

42 minimum_version: Literal["1.2", "1.3"] = "1.2" 

43 ciphers: str | None = None 

44 

45 @model_validator(mode="after") 

46 def _check_verify_source(self): 

47 if self.require_client_cert and self.ca_certs is None: 

48 raise ValueError("require_client_cert needs 'ca_certs' to verify clients against") 

49 

50 return self 

51 

52 def uvicorn_options(self) -> dict[str, Any]: 

53 """Return the ssl keyword arguments for a uvicorn config. 

54 

55 Raises: 

56 SecurityConfigError: The named password variable is unset or empty. 

57 """ 

58 password = None 

59 

60 if self.keyfile_password_env: 

61 password = os.environ.get(self.keyfile_password_env) 

62 

63 if not password: 

64 raise SecurityConfigError( 

65 f"env var {self.keyfile_password_env!r} holds no private key password" 

66 ) 

67 

68 options = { 

69 "ssl_certfile": self.certfile, 

70 "ssl_keyfile": self.keyfile, 

71 "ssl_keyfile_password": password, 

72 "ssl_ca_certs": str(self.ca_certs) if self.ca_certs else None, 

73 "ssl_cert_reqs": ssl.CERT_REQUIRED if self.require_client_cert else ssl.CERT_NONE, 

74 } 

75 

76 if self.ciphers: 

77 options["ssl_ciphers"] = self.ciphers 

78 

79 return options 

80 

81 def apply_minimum_version(self, context: ssl.SSLContext): 

82 """Raise the floor on the protocol versions the listener will negotiate. 

83 

84 The server builds its own context from the uvicorn options, and the version 

85 floor is not one of them, so it is set on the finished context. 

86 """ 

87 context.minimum_version = TLS_VERSIONS[self.minimum_version] 

88 

89 

90class Authenticator(ABC): 

91 """Decides whether a request carries acceptable credentials.""" 

92 

93 @property 

94 @abstractmethod 

95 def enabled(self) -> bool: 

96 """Whether any credential is required at all.""" 

97 

98 @abstractmethod 

99 def authorize(self, method: str, path: str, authorization: str | None) -> bool: 

100 """Return whether a request for `path` using `method` and `authorization` may proceed.""" 

101 

102 def openapi_scheme(self) -> dict[str, Any] | None: 

103 """Return the OpenAPI security scheme to advertise, where one applies.""" 

104 return None 

105 

106 

107class OpenAccess(Authenticator): 

108 """Accepts every request.""" 

109 

110 @property 

111 @override 

112 def enabled(self): 

113 return False 

114 

115 @override 

116 def authorize(self, method: str, path: str, authorization: str | None): 

117 return True 

118 

119 

120class BearerToken(Authenticator): 

121 """Accepts requests presenting a shared bearer token.""" 

122 

123 def __init__(self, token: str, *, allow_anonymous_read: bool): 

124 self.token = token 

125 self.allow_anonymous_read = allow_anonymous_read 

126 

127 @property 

128 @override 

129 def enabled(self): 

130 return True 

131 

132 @override 

133 def authorize(self, method: str, path: str, authorization: str | None): 

134 # The schema and the documentation UIs map out the endpoints an anonymous reader 

135 # is not allowed to reach, so they stay behind the token. 

136 if self.allow_anonymous_read and method in READ_METHODS and path not in DOC_PATHS: 

137 return True 

138 

139 if authorization is None or not authorization.startswith(BEARER_PREFIX): 

140 return False 

141 

142 return secrets.compare_digest(authorization[len(BEARER_PREFIX) :], self.token) 

143 

144 @override 

145 def openapi_scheme(self): 

146 return {"type": "http", "scheme": "bearer"} 

147 

148 

149class NoAuthConfig(BaseModel): 

150 """Leave the web API open to anyone who can reach the port.""" 

151 

152 kind: Literal["none"] = "none" 

153 

154 def create_authenticator(self) -> Authenticator: 

155 return OpenAccess() 

156 

157 

158class TokenAuthConfig(BaseModel): 

159 """Require a shared bearer token on requests to the web API. 

160 

161 The token itself is never part of this record, since the key-value store holding 

162 it is readable across the bus. Name an environment variable or a file instead. 

163 """ 

164 

165 kind: Literal["token"] = "token" 

166 token_env: str | None = None 

167 token_file: pathlib.Path | None = None 

168 allow_anonymous_read: bool = False 

169 

170 @model_validator(mode="after") 

171 def _check_one_source(self): 

172 if (self.token_env is None) == (self.token_file is None): 

173 raise ValueError("token auth needs exactly one of 'token_env' or 'token_file'") 

174 

175 return self 

176 

177 def create_authenticator(self) -> Authenticator: 

178 """Build the authenticator, reading the token from its configured source. 

179 

180 Raises: 

181 SecurityConfigError: The source is unreadable or holds nothing. 

182 """ 

183 if self.token_env is not None: 

184 token = os.environ.get(self.token_env, "") 

185 source = f"env var {self.token_env!r}" 

186 else: 

187 assert self.token_file is not None 

188 

189 try: 

190 token = self.token_file.read_text(encoding="utf-8").strip() 

191 except OSError as err: 

192 raise SecurityConfigError( 

193 f"cannot read web API token file {self.token_file}" 

194 ) from err 

195 

196 source = f"token file {self.token_file}" 

197 

198 if not token: 

199 raise SecurityConfigError(f"web API {source} holds no token") 

200 

201 if len(token) < SHORT_TOKEN_LENGTH: 

202 logger.warning( 

203 f"web API token from {source} is shorter than {SHORT_TOKEN_LENGTH} characters " 

204 f"and is weak against guessing" 

205 ) 

206 

207 return BearerToken(token, allow_anonymous_read=self.allow_anonymous_read) 

208 

209 

210type AuthConfig = Annotated[NoAuthConfig | TokenAuthConfig, Field(discriminator="kind")] 

211 

212 

213class CORSConfig(BaseModel): 

214 """Cross-origin rules for browser clients. 

215 

216 No origin is permitted by default. A dashboard served from somewhere other than 

217 the web API's own origin needs listing here. 

218 """ 

219 

220 allow_origins: list[str] = [] 

221 allow_origin_regex: str | None = None 

222 allow_credentials: bool = False 

223 allow_methods: list[str] = ["*"] 

224 allow_headers: list[str] = ["*"] 

225 

226 @model_validator(mode="after") 

227 def _check_credentialed_wildcard(self): 

228 # Browsers reject a wildcard origin on a credentialed response, so the 

229 # combination would silently deny every request it appears to allow. 

230 if self.allow_credentials and "*" in self.allow_origins: 

231 raise ValueError("allow_credentials cannot be used with a wildcard origin") 

232 

233 return self 

234 

235 def middleware_options(self) -> dict[str, Any]: 

236 """Return the keyword arguments for the CORS middleware.""" 

237 return { 

238 "allow_origins": self.allow_origins, 

239 "allow_origin_regex": self.allow_origin_regex, 

240 "allow_credentials": self.allow_credentials, 

241 "allow_methods": self.allow_methods, 

242 "allow_headers": self.allow_headers, 

243 } 

244 

245 

246class AuthMiddleware: 

247 """Rejects requests the authenticator does not permit, before routing. 

248 

249 Enforcing here rather than per route means an endpoint added later is covered 

250 without having to remember a dependency. 

251 """ 

252 

253 def __init__(self, app, authenticator: Authenticator): 

254 self.app = app 

255 self.authenticator = authenticator 

256 

257 async def __call__(self, scope, receive, send): 

258 if scope["type"] != "http": 

259 await self.app(scope, receive, send) 

260 return 

261 

262 authorization = next( 

263 (value.decode("latin-1") for key, value in scope["headers"] if key == b"authorization"), 

264 None, 

265 ) 

266 

267 if not self.authenticator.authorize(scope["method"], scope["path"], authorization): 

268 response = JSONResponse( 

269 {"detail": "Not authenticated"}, 

270 status_code=401, 

271 headers={"WWW-Authenticate": "Bearer"}, 

272 ) 

273 await response(scope, receive, send) 

274 return 

275 

276 await self.app(scope, receive, send) 

277 

278 

279class SecurityHeadersMiddleware: 

280 """Adds response headers that constrain how browsers treat the API.""" 

281 

282 def __init__(self, app, *, hsts: bool): 

283 self.app = app 

284 self.hsts = hsts 

285 

286 async def __call__(self, scope, receive, send): 

287 if scope["type"] != "http": 

288 await self.app(scope, receive, send) 

289 return 

290 

291 extra = [(b"x-content-type-options", b"nosniff")] 

292 

293 if self.hsts: 

294 extra.append((b"strict-transport-security", b"max-age=31536000")) 

295 

296 async def send_with_headers(message): 

297 if message["type"] == "http.response.start": 

298 message["headers"] = [*message.get("headers", []), *extra] 

299 

300 await send(message) 

301 

302 await self.app(scope, receive, send_with_headers)