Coverage for espie_character_gen/server.py: 66%

1import os 

2from typing import Type, TypeVar 

3from fastapi import Depends, FastAPI, HTTPException, Request 

4from importlib.metadata import version 

5 

6from fastapi.responses import JSONResponse, RedirectResponse 

7 

8from fastapi.middleware.cors import CORSMiddleware 

9from fastapi.templating import Jinja2Templates 

10from pydantic import BaseModel 

11from espie_character_gen import configs 

12from espie_character_gen.rest_app.authentication import ( 

13 get_verifier, 

14 token_auth_scheme, 

15) 

16from fastapi_sessions.frontends.implementations import SessionCookie, CookieParameters 

17from uuid import UUID, uuid4 

18from fastapi_sessions.backends.implementations import InMemoryBackend 

19 

20from espie_character_gen.rest_app.model import AppSession 

21from fastapi_sessions.session_verifier import SessionVerifier 

22from fastapi import HTTPException 

23 

24 

25class BasicVerifier(SessionVerifier[UUID, AppSession]): 

26 def __init__( 

27 self, 

28 *, 

29 identifier: str, 

30 auto_error: bool, 

31 backend: InMemoryBackend[UUID, AppSession], 

32 auth_http_exception: HTTPException, 

33 ): 

34 self._identifier = identifier 

35 self._auto_error = auto_error 

36 self._backend = backend 

37 self._auth_http_exception = auth_http_exception 

38 

39 @property 

40 def identifier(self): 

41 return self._identifier 

42 

43 @property 

44 def backend(self): 

45 return self._backend 

46 

47 @property 

48 def auto_error(self): 

49 return self._auto_error 

50 

51 @property 

52 def auth_http_exception(self): 

53 return self._auth_http_exception 

54 

55 def verify_session(self, model: AppSession) -> bool: 

56 """If the session exists, it is valid""" 

57 return True 

58 

59 

60backend = InMemoryBackend[UUID, AppSession]() 

61verifier = BasicVerifier( 

62 identifier="general_verifier", 

63 auto_error=False, 

64 backend=backend, 

65 auth_http_exception=HTTPException(status_code=403, detail="invalid session"), 

66) 

67 

68 

69cookie_params = CookieParameters() 

70 

71# Uses UUID 

72cookie = SessionCookie( 

73 cookie_name="pookie", 

74 identifier="general_verifier", 

75 auto_error=False, 

76 secret_key=configs.secret_key, 

77 cookie_params=cookie_params, 

78) 

79 

80try: 

81 _version = version("espie_character_gen") 

82except: # pragma: no cover 

83 _version = "dev" 

84 

85fastapi_app = FastAPI( 

86 title="Espie Character Gen API", 

87 version=_version, 

88) 

89 

90fastapi_app.add_middleware( 

91 CORSMiddleware, 

92 allow_origins=configs.cors_origins, 

93 allow_credentials=configs.cors_allow_credentials, 

94 allow_methods=configs.cors_methods, 

95 allow_headers=configs.cors_headers, 

96) 

97 

98 

99@fastapi_app.middleware("http") 

100async def hydrate_user( 

101 request: Request, 

102 call_next, 

103): 

104 try: 

105 token = await token_auth_scheme(request) 

106 except HTTPException as e: 

107 if e.detail == "Not authenticated": 

108 return await call_next(request) 

109 return JSONResponse( 

110 {"status": "error", "message": "Not Authenticated"}, status_code=401 

111 ) 

112 

113 result = get_verifier().verify(token.credentials) 

114 

115 if result.get("status"): 

116 return JSONResponse(result, status_code=403) 

117 request.state.user = result 

118 return await call_next(request) 

119 

120 

121@fastapi_app.get("/healthz") 

122def healthz(): 

123 return {"o": "k"} 

124 

125 

126@fastapi_app.get("/version") 

127def version(): 

128 return {"version": _version} 

129 

130 

131@fastapi_app.get("/me") 

132def get_me(request: Request): 

133 return getattr(request.state, "user", None) 

134 

135 

136templates = Jinja2Templates( 

137 directory=os.path.join(os.path.dirname(__file__), "templates") 

138) 

139 

140 

141def link(url: str, name: str): 

142 return {"url": url, "name": name} 

143 

144 

145SHARED_LINKS = [ 

146 link("/", "Home"), 

147 link("/about", "About"), 

148 link("/contact", "Contact"), 

149] 

150 

151LOGGED_OUT_LINKS = [ 

152 *SHARED_LINKS, 

153 link("/login", "Login"), 

154] 

155 

156LOGGED_IN_LINKS = [ 

157 *SHARED_LINKS, 

158 link("/logout", "Logout"), 

159] 

160 

161 

162def context(*, is_logged_in: bool = False, **kwargs): 

163 return { 

164 **kwargs, 

165 "links": LOGGED_IN_LINKS if is_logged_in else LOGGED_OUT_LINKS, 

166 } 

167 

168 

169@fastapi_app.get("/", dependencies=[Depends(cookie)]) 

170def index(request: Request, data: AppSession = Depends(verifier)): 

171 print(data) 

172 return templates.TemplateResponse( 

173 request=request, 

174 name="index.html", 

175 context=context( 

176 name=getattr(data, "username", "Billy Ichiban"), 

177 is_logged_in=data is not None, 

178 ), 

179 ) 

180 

181 

182class LoginCreds(BaseModel): 

183 username: str 

184 password: str 

185 

186 

187_TModel = TypeVar("_TModel", bound=BaseModel) 

188 

189 

190def form_or_json(model: Type[_TModel]) -> _TModel: 

191 async def form_or_json_inner(request: Request) -> _TModel: 

192 type_ = request.headers["Content-Type"].split(";", 1)[0] 

193 if type_ == "application/json": 

194 data = await request.json() 

195 elif type_ == "application/x-www-form-urlencoded": 

196 data = await request.form() 

197 else: 

198 raise HTTPException(400) 

199 return model.model_validate(data) 

200 

201 return Depends(form_or_json_inner) 

202 

203 

204@fastapi_app.get("/login") 

205def login(request: Request): 

206 return templates.TemplateResponse( 

207 request=request, 

208 name="login.html", 

209 context=context(is_logged_in=False), 

210 ) 

211 

212 

213@fastapi_app.post("/login") 

214async def login(request: Request, login_request: LoginCreds = form_or_json(LoginCreds)): 

215 if not login_request.password or login_request.password == "failure": 

216 raise HTTPException(status_code=401, detail="Invalid Credentials") 

217 session = uuid4() 

218 data = AppSession( 

219 username=login_request.username, 

220 ) 

221 await backend.create(session, data) 

222 

223 if request.headers["Content-Type"] == "application/json": 

224 response = JSONResponse({"session": str(session)}) 

225 else: 

226 response = RedirectResponse("/", status_code=302) 

227 

228 cookie.attach_to_response(response, session) 

229 return response 

230 

231 

232@fastapi_app.get("/logout") 

233async def logout(session_id: UUID = Depends(cookie)): 

234 response = RedirectResponse("/", status_code=302) 

235 await backend.delete(session_id) 

236 cookie.delete_from_response(response) 

237 return response 

238 

239 

240@fastapi_app.get("/{path:path}") 

241def catch_all(path: str): 

242 return RedirectResponse(url="/", status_code=302)