96 lines
2.6 KiB
Python
96 lines
2.6 KiB
Python
from http.cookies import SimpleCookie
|
|
|
|
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,
|
|
ServiceError,
|
|
)
|
|
from joserfc.errors import JoseError
|
|
from layercake.dynamodb import DynamoDBPersistenceLayer, KeyPair, SortKey
|
|
|
|
from boto3clients import dynamodb_client
|
|
from config import OAUTH2_TABLE
|
|
from oauth2 import server
|
|
|
|
router = Router()
|
|
logger = Logger(__name__)
|
|
dyn = DynamoDBPersistenceLayer(OAUTH2_TABLE, dynamodb_client)
|
|
|
|
|
|
@router.get('/authorize')
|
|
def authorize():
|
|
current_event = router.current_event
|
|
cookies = _parse_cookies(current_event.get('cookies', []))
|
|
session = cookies.get('__session')
|
|
|
|
if not session:
|
|
raise BadRequestError('Missing session')
|
|
|
|
try:
|
|
sid, sub = session.split(':')
|
|
# Raise if session is not found
|
|
dyn.collection.get_item(
|
|
KeyPair('SESSION', sid),
|
|
exc_cls=InvalidSession,
|
|
)
|
|
grant = server.get_consent_grant(
|
|
request=router.current_event,
|
|
end_user=sub,
|
|
)
|
|
user_scopes = _user_scopes(sub)
|
|
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')
|
|
|
|
return server.create_authorization_response(
|
|
request=router.current_event,
|
|
grant_user=sub,
|
|
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
|
|
)
|
|
|
|
|
|
def _user_scopes(sub: str) -> set:
|
|
return set(
|
|
scope_to_list(
|
|
dyn.collection.get_item(
|
|
KeyPair(
|
|
pk=sub,
|
|
sk=SortKey(sk='SCOPE', path_spec='scope'),
|
|
),
|
|
exc_cls=BadRequestError,
|
|
)
|
|
)
|
|
)
|
|
|
|
|
|
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 InvalidSession(BadRequestError): ...
|