100 lines
2.9 KiB
Python
100 lines
2.9 KiB
Python
from http import HTTPStatus
|
|
from http.cookies import SimpleCookie
|
|
|
|
import jwt
|
|
from authlib.oauth2.rfc6749 import errors
|
|
from authlib.oauth2.rfc6749.util import scope_to_list
|
|
from aws_lambda_powertools import Logger
|
|
from aws_lambda_powertools.event_handler.api_gateway import Router
|
|
from aws_lambda_powertools.event_handler.exceptions import BadRequestError, ServiceError
|
|
from layercake.dynamodb import DynamoDBPersistenceLayer, KeyPair
|
|
|
|
from boto3clients import dynamodb_client
|
|
from config import ISSUER, JWT_ALGORITHM, JWT_SECRET, OAUTH2_TABLE
|
|
from oauth2 import server
|
|
|
|
router = Router()
|
|
logger = Logger(__name__)
|
|
oauth2_layer = DynamoDBPersistenceLayer(OAUTH2_TABLE, dynamodb_client)
|
|
|
|
|
|
@router.get('/authorize')
|
|
def authorize():
|
|
current_event = router.current_event
|
|
cookies = _parse_cookies(current_event.get('cookies', []))
|
|
session_id = cookies.get('session_id')
|
|
|
|
if not session_id:
|
|
raise BadRequestError('Missing session_id')
|
|
|
|
try:
|
|
sub, session_scope = verify_session(session_id)
|
|
grant = server.get_consent_grant(
|
|
request=router.current_event,
|
|
end_user=sub,
|
|
)
|
|
client_scopes = set(scope_to_list(grant.client.scope))
|
|
user_scopes = set(scope_to_list(session_scope)) if session_scope else set()
|
|
|
|
if not client_scopes.issubset(
|
|
user_scopes | {'openid', 'email', 'profile', 'offline_access'}
|
|
):
|
|
raise errors.InvalidScopeError(status_code=HTTPStatus.UNAUTHORIZED)
|
|
|
|
return server.create_authorization_response(
|
|
request=router.current_event,
|
|
grant_user=sub,
|
|
grant=grant,
|
|
)
|
|
except jwt.exceptions.InvalidTokenError as err:
|
|
logger.exception(err)
|
|
raise BadRequestError(str(err))
|
|
except errors.OAuth2Error as err:
|
|
logger.exception(err)
|
|
raise ServiceError(
|
|
status_code=err.status_code,
|
|
msg=dict(err.get_body()), # type: ignore
|
|
)
|
|
|
|
|
|
def verify_session(session_id: str) -> tuple[str, str | None]:
|
|
payload = jwt.decode(
|
|
session_id,
|
|
JWT_SECRET,
|
|
algorithms=[JWT_ALGORITHM],
|
|
issuer=ISSUER,
|
|
options={
|
|
'require': ['exp', 'sub', 'iss', 'sid'],
|
|
'leeway': 60,
|
|
},
|
|
)
|
|
|
|
oauth2_layer.collection.get_item(
|
|
KeyPair(
|
|
pk='SESSION',
|
|
sk=payload['sid'],
|
|
),
|
|
exc_cls=SessionRevokedError,
|
|
)
|
|
|
|
return payload['sub'], payload.get('scope')
|
|
|
|
|
|
def _parse_cookies(cookies: list[str] | None) -> dict[str, str]:
|
|
parsed_cookies = {}
|
|
|
|
if not cookies:
|
|
return parsed_cookies
|
|
|
|
for s in cookies:
|
|
c = SimpleCookie()
|
|
c.load(s)
|
|
parsed_cookies.update({k: morsel.value for k, morsel in c.items()})
|
|
|
|
return parsed_cookies
|
|
|
|
|
|
class SessionRevokedError(BadRequestError):
|
|
def __init__(self, *_):
|
|
super().__init__('Session revoked')
|