-
Notifications
You must be signed in to change notification settings - Fork 1
/
session.py
81 lines (68 loc) · 2.89 KB
/
session.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
"""
Supplies dependencies needed for session resolution.
"""
from typing import Optional, Tuple
import redis
from headers import AUTHORIZATION
from starlette import status
from starlette.applications import ASGIApp
from starlette.middleware.authentication import (
AuthCredentials,
AuthenticationBackend,
AuthenticationError,
AuthenticationMiddleware,
)
from starlette.requests import HTTPConnection, Request
from ..errors import ErrorResponse, ApiError
from ..security import User, scopes, SessionManager
class SessionMiddleware(AuthenticationMiddleware, AuthenticationBackend):
manager: SessionManager
@staticmethod
def get(r: Request) -> SessionManager:
return r.state.sessions
@staticmethod
def user(r: Request) -> User:
return r.user
def __init__(self, app: ASGIApp, url: str, token_bytes: int, lifetime: int):
super(SessionMiddleware, self).__init__(app, backend=self, on_error=SessionMiddleware.on_error)
self.redis = redis.from_url(url)
self.manager = SessionManager(redis=self.redis, token_bytes=token_bytes, lifetime=lifetime)
def __del__(self):
self.redis.close()
@staticmethod
def on_error(_: HTTPConnection, exc: AuthenticationError):
"""Customizes the authentication errors
"""
try:
message = exc.__cause__.args[0]
except (AttributeError, IndexError):
message = "Error in Authentication"
return ErrorResponse(error=ApiError(code=status.HTTP_401_UNAUTHORIZED, message=message))
async def authenticate(self, conn: HTTPConnection) -> Optional[Tuple[AuthCredentials, User]]:
"""Extracts Session data from requests
Adds a property to the request state to access the SessionManager
r.state.sessions: SessionManager
"""
conn.state.sessions = self.manager
header = conn.headers.get(AUTHORIZATION, None)
if header is not None:
scheme, _, credentials = header.partition(" ")
if scheme.lower() != "bearer":
raise AuthenticationError() from ValueError("Wrong Scheme")
try:
session = self.manager.get_session(credentials)
user = User.from_cache(username=session.user, token=credentials)
session_data = session.data
if "projects" in session_data:
user.admin_projects = set(session_data["projects"])
creds = AuthCredentials(scopes.AUTHENTICATED)
conn.scope["session"] = session_data
if "scopes" in session_data:
creds.scopes.append(session_data["scopes"])
user.scopes.update(session_data["scopes"])
except ValueError as e:
raise AuthenticationError from e
else:
user = User.null()
creds = AuthCredentials()
return creds, user