darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2025-12-10 23:32:27 -05:00
Code Issues Packages Projects Releases Wiki Activity

try another way

Browse Source
This commit is contained in:
Kenneth Kehl
2024-09-05 09:16:32 -07:00
parent 33568d431f
commit befe20249f
1 changed files with 30 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
app/authentication/auth.py
View File

@@ -1,3 +1,4 @@
import os
import uuid
from flask import current_app, g, request
@@ -15,7 +16,6 @@ from notifications_python_client.errors import (
from sqlalchemy.orm.exc import NoResultFound
from app.serialised_models import SerialisedService
from app.utils import debug_not_production
from notifications_utils import request_helper
# stvnrlly - this is silly, but bandit has a multiline string bug (https://github.com/PyCQA/bandit/issues/658)
@@ -63,36 +63,30 @@ def requires_admin_auth():
def requires_internal_auth(expected_client_id):
debug_not_production(
f"TODO REMOVE: Enter requires_internal_auth with expected client id {expected_client_id}"
)
# Looks like we are hitting this for some reason
# expected_client_id looks like ADMIN_CLIENT_USERNAME on the admin side, and
# INTERNAL_CLIENT_API_KEYS is a dict
keys = current_app.config.get("INTERNAL_CLIENT_API_KEYS")
if keys.get(expected_client_id) is None:
debug_not_production(
f"TODO REMOVE: {expected_client_id} not in {keys}, raising TypeError\n"
)
raise TypeError("Unknown client_id for internal auth")
err_msg = "Unknown client_id for internal auth"
current_app.logger.error(err_msg)
raise TypeError(err_msg)
request_helper.check_proxy_header_before_request()
auth_token = _get_auth_token(request)
debug_not_production(f"TODO REMOVE: auth token {auth_token}")
client_id = _get_token_issuer(auth_token)
debug_not_production(f"TODO_REMOVE: client id {client_id}")
if client_id != expected_client_id:
current_app.logger.info("client_id: %s", client_id)
current_app.logger.info("expected_client_id: %s", expected_client_id)
raise AuthError("Unauthorized: not allowed to perform this action", 401)
err_msg = "Unauthorized: not allowed to perform this action"
current_app.logger.error(err_msg)
raise AuthError(err_msg, 401)
api_keys = [
InternalApiKey(client_id, secret)
for secret in current_app.config.get("INTERNAL_CLIENT_API_KEYS")[client_id]
]
debug_not_production(
f"got the api keys ... note client_id {client_id} should be the service_id {api_keys}"
)
_decode_jwt_token(auth_token, api_keys, client_id)
g.service_id = client_id
@@ -140,13 +134,19 @@ def requires_auth():
def _decode_jwt_token(auth_token, api_keys, service_id=None):
# Temporary expedient to get e2e tests working. If we are in
# the development or staging environments, just return the first
# api key.
if os.getenv("NOTIFY_ENVIRONMENT") in ["development", "staging"]:
for api_key in api_keys:
return api_key
for api_key in api_keys:
try:
decode_jwt_token(auth_token, api_key.secret)
except TypeError:
debug_not_production(
f"TODO REMOVE: Hit TypeError!!! service_id {service_id} api_keys {api_keys}"
)
err_msg = "Invalid token: type error"
current_app.logger.exception(err_msg)
raise AuthError(
"Invalid token: type error",
403,
@@ -158,11 +158,13 @@ def _decode_jwt_token(auth_token, api_keys, service_id=None):
err_msg = (
"Error: Your system clock must be accurate to within 30 seconds"
)
current_app.logger.exception(err_msg)
raise AuthError(
err_msg, 403, service_id=service_id, api_key_id=api_key.id
)
except TokenAlgorithmError:
err_msg = "Invalid token: algorithm used is not HS256"
current_app.logger.exception(err_msg)
raise AuthError(err_msg, 403, service_id=service_id, api_key_id=api_key.id)
except TokenDecodeError:
# we attempted to validate the token but it failed meaning it was not signed using this api key.
@@ -170,8 +172,12 @@ def _decode_jwt_token(auth_token, api_keys, service_id=None):
# TODO: Change this so it doesn't also catch `TokenIssuerError` or `TokenIssuedAtError` exceptions (which
# are children of `TokenDecodeError`) as these should cause an auth error immediately rather than
# continue on to check the next API key
current_app.logger.exception(
"TokenDecodeError. Couldn't decode auth token for given api key"
)
continue
except TokenError:
current_app.logger.exception("TokenError")
# General error when trying to decode and validate the token
raise AuthError(
GENERAL_TOKEN_ERROR_MESSAGE,
@@ -181,8 +187,10 @@ def _decode_jwt_token(auth_token, api_keys, service_id=None):
)
if api_key.expiry_date:
err_msg = "Invalid token: API key revoked"
current_app.logger.error(err_msg, exc_info=True)
raise AuthError(
"Invalid token: API key revoked",
err_msg,
403,
service_id=service_id,
api_key_id=api_key.id,
@@ -190,13 +198,11 @@ def _decode_jwt_token(auth_token, api_keys, service_id=None):
return api_key
else:
# We are hitting this in the e2e tests, debug why
debug_not_production(
f"auth token {auth_token} keys {api_keys} service_id={service_id}"
)
# service has API keys, but none matching the one the user provided
raise AuthError("Invalid token: API key not found", 403, service_id=service_id)
# if we get here, we probably hit TokenDecodeErrors earlier
err_msg = "Invalid token: API key not found"
current_app.logger.error(err_msg, exc_info=True)
raise AuthError(err_msg, 403, service_id=service_id)
def _get_auth_token(req):