88 lines
2.3 KiB
Python
88 lines
2.3 KiB
Python
from http.cookies import SimpleCookie
|
|
|
|
import jwt
|
|
from authlib.oauth2.rfc6749 import errors
|
|
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
|
|
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:
|
|
user_id = verify_session(session_id)
|
|
grant = server.get_consent_grant(
|
|
request=router.current_event,
|
|
end_user={'id': user_id},
|
|
)
|
|
|
|
return server.create_authorization_response(
|
|
request=router.current_event,
|
|
grant_user={'id': user_id},
|
|
grant=grant,
|
|
)
|
|
except jwt.exceptions.InvalidTokenError as err:
|
|
logger.exception(err)
|
|
raise BadRequestError(str(err))
|
|
except errors.OAuth2Error as err:
|
|
logger.exception(err)
|
|
return dict(err.get_body())
|
|
|
|
|
|
def verify_session(session_id: str) -> str:
|
|
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']
|
|
|
|
|
|
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')
|