70 lines
2.1 KiB
Python
70 lines
2.1 KiB
Python
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,
|
|
ForbiddenError,
|
|
NotFoundError,
|
|
ServiceError,
|
|
)
|
|
from joserfc.errors import JoseError
|
|
from layercake.dynamodb import DynamoDBPersistenceLayer, KeyPair
|
|
|
|
from boto3clients import dynamodb_client
|
|
from config import USER_TABLE
|
|
from oauth2 import get_user_scope, server
|
|
from util import parse_cookies
|
|
|
|
router = Router()
|
|
logger = Logger(__name__)
|
|
dyn = DynamoDBPersistenceLayer(USER_TABLE, dynamodb_client)
|
|
|
|
|
|
class SessionNotFoundError(NotFoundError): ...
|
|
|
|
|
|
@router.get('/authorize')
|
|
def authorize():
|
|
current_event = router.current_event
|
|
cookies = parse_cookies(current_event.get('cookies', []))
|
|
session = cookies.get('SID')
|
|
|
|
if not session:
|
|
raise BadRequestError('Session cookie (SID) is required')
|
|
|
|
try:
|
|
session_id, user_id = session.split(':')
|
|
# Raise if session is not found
|
|
dyn.collection.get_item(
|
|
KeyPair('SESSION', session_id),
|
|
exc_cls=SessionNotFoundError,
|
|
)
|
|
grant = server.get_consent_grant(
|
|
request=router.current_event,
|
|
end_user=user_id,
|
|
)
|
|
user_scopes = get_user_scope(user_id)
|
|
client_scopes = set(scope_to_list(grant.client.scope))
|
|
|
|
# Deny authorization if user lacks scopes requested by client
|
|
if not client_scopes.issubset(user_scopes):
|
|
raise ForbiddenError('Access denied')
|
|
|
|
response = server.create_authorization_response(
|
|
request=router.current_event,
|
|
grant_user=user_id,
|
|
grant=grant,
|
|
)
|
|
except JoseError 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
|
|
)
|
|
else:
|
|
return response
|