mirror of
https://github.com/GSA/notifications-api.git
synced 2026-01-03 12:30:44 -05:00
Merge pull request #3300 from alphagov/multi-internal-auth-179039225
Split up authentication for internal API endpoint
This commit is contained in:
Ben Thorner
@@ -271,10 +271,16 @@ def register_blueprint(application):
|
||||
|
||||
|
||||
def register_v2_blueprints(application):
|
||||
from app.authentication.auth import requires_auth
|
||||
from app.authentication.auth import (
|
||||
requires_auth,
|
||||
requires_govuk_alerts_auth,
|
||||
)
|
||||
from app.v2.broadcast.post_broadcast import (
|
||||
v2_broadcast_blueprint as post_broadcast,
|
||||
)
|
||||
from app.v2.govuk_alerts.get_broadcasts import (
|
||||
v2_govuk_alerts_blueprint as get_broadcasts,
|
||||
)
|
||||
from app.v2.inbound_sms.get_inbound_sms import (
|
||||
v2_inbound_sms_blueprint as get_inbound_sms,
|
||||
)
|
||||
@@ -315,6 +321,9 @@ def register_v2_blueprints(application):
|
||||
post_broadcast.before_request(requires_auth)
|
||||
application.register_blueprint(post_broadcast)
|
||||
|
||||
get_broadcasts.before_request(requires_govuk_alerts_auth)
|
||||
application.register_blueprint(get_broadcasts)
|
||||
|
||||
|
||||
def init_app(app):
|
||||
|
||||
|
||||
@@ -49,55 +49,50 @@ class AuthError(Exception):
|
||||
}
|
||||
|
||||
|
||||
def get_auth_token(req):
|
||||
auth_header = req.headers.get('Authorization', None)
|
||||
if not auth_header:
|
||||
raise AuthError('Unauthorized: authentication token must be provided', 401)
|
||||
|
||||
auth_scheme = auth_header[:7].title()
|
||||
|
||||
if auth_scheme != 'Bearer ':
|
||||
raise AuthError('Unauthorized: authentication bearer scheme must be used', 401)
|
||||
|
||||
return auth_header[7:]
|
||||
class InternalApiKey():
|
||||
def __init__(self, client_id, secret):
|
||||
self.secret = secret
|
||||
self.id = client_id
|
||||
self.expiry_date = None
|
||||
|
||||
|
||||
def requires_no_auth():
|
||||
pass
|
||||
|
||||
|
||||
def requires_govuk_alerts_auth():
|
||||
requires_internal_auth(current_app.config.get('GOVUK_ALERTS_CLIENT_ID'))
|
||||
|
||||
|
||||
def requires_admin_auth():
|
||||
requires_internal_auth(current_app.config.get('ADMIN_CLIENT_ID'))
|
||||
|
||||
|
||||
def requires_internal_auth(expected_client_id):
|
||||
if expected_client_id not in current_app.config.get('INTERNAL_CLIENT_API_KEYS'):
|
||||
raise TypeError("Unknown client_id for internal auth")
|
||||
|
||||
request_helper.check_proxy_header_before_request()
|
||||
auth_token = _get_auth_token(request)
|
||||
client_id = _get_token_issuer(auth_token)
|
||||
|
||||
auth_token = get_auth_token(request)
|
||||
client = __get_token_issuer(auth_token)
|
||||
if client_id != expected_client_id:
|
||||
raise AuthError("Unauthorized: not allowed to perform this action", 401)
|
||||
|
||||
if client == current_app.config.get('ADMIN_CLIENT_USER_NAME'):
|
||||
g.service_id = current_app.config.get('ADMIN_CLIENT_USER_NAME')
|
||||
api_keys = [
|
||||
InternalApiKey(client_id, secret)
|
||||
for secret in current_app.config.get('INTERNAL_CLIENT_API_KEYS')[client_id]
|
||||
]
|
||||
|
||||
for secret in current_app.config.get('API_INTERNAL_SECRETS'):
|
||||
try:
|
||||
decode_jwt_token(auth_token, secret)
|
||||
return
|
||||
except TokenExpiredError:
|
||||
raise AuthError("Invalid token: expired, check that your system clock is accurate", 403)
|
||||
except TokenDecodeError:
|
||||
# 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 admin client secret
|
||||
continue
|
||||
|
||||
# Either there are no admin client secrets or their token didn't match one of them so error
|
||||
raise AuthError("Unauthorized: admin authentication token not found", 401)
|
||||
else:
|
||||
raise AuthError('Unauthorized: admin authentication token required', 401)
|
||||
_decode_jwt_token(auth_token, api_keys, client_id)
|
||||
g.service_id = client_id
|
||||
|
||||
|
||||
def requires_auth():
|
||||
request_helper.check_proxy_header_before_request()
|
||||
|
||||
auth_token = get_auth_token(request)
|
||||
issuer = __get_token_issuer(auth_token) # ie the `iss` claim which should be a service ID
|
||||
auth_token = _get_auth_token(request)
|
||||
issuer = _get_token_issuer(auth_token) # ie the `iss` claim which should be a service ID
|
||||
|
||||
try:
|
||||
service_id = uuid.UUID(issuer)
|
||||
@@ -116,15 +111,23 @@ def requires_auth():
|
||||
if not service.active:
|
||||
raise AuthError("Invalid token: service is archived", 403, service_id=service.id)
|
||||
|
||||
for api_key in service.api_keys:
|
||||
api_key = _decode_jwt_token(auth_token, service.api_keys, service.id)
|
||||
|
||||
g.api_user = api_key
|
||||
g.service_id = service_id
|
||||
g.authenticated_service = service
|
||||
|
||||
|
||||
def _decode_jwt_token(auth_token, api_keys, service_id=None):
|
||||
for api_key in api_keys:
|
||||
try:
|
||||
decode_jwt_token(auth_token, api_key.secret)
|
||||
except TokenExpiredError:
|
||||
err_msg = "Error: Your system clock must be accurate to within 30 seconds"
|
||||
raise AuthError(err_msg, 403, service_id=service.id, api_key_id=api_key.id)
|
||||
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"
|
||||
raise AuthError(err_msg, 403, service_id=service.id, api_key_id=api_key.id)
|
||||
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.
|
||||
# Let's try the next one
|
||||
@@ -134,28 +137,38 @@ def requires_auth():
|
||||
continue
|
||||
except TokenError:
|
||||
# General error when trying to decode and validate the token
|
||||
raise AuthError(GENERAL_TOKEN_ERROR_MESSAGE, 403, service_id=service.id, api_key_id=api_key.id)
|
||||
raise AuthError(GENERAL_TOKEN_ERROR_MESSAGE, 403, service_id=service_id, api_key_id=api_key.id)
|
||||
|
||||
if api_key.expiry_date:
|
||||
raise AuthError("Invalid token: API key revoked", 403, service_id=service.id, api_key_id=api_key.id)
|
||||
|
||||
g.service_id = service.id
|
||||
g.api_user = api_key
|
||||
g.authenticated_service = service
|
||||
raise AuthError("Invalid token: API key revoked", 403, service_id=service_id, api_key_id=api_key.id)
|
||||
|
||||
current_app.logger.info('API authorised for service {} with api key {}, using issuer {} for URL: {}'.format(
|
||||
service.id,
|
||||
service_id,
|
||||
api_key.id,
|
||||
request.headers.get('User-Agent'),
|
||||
request.base_url
|
||||
))
|
||||
return
|
||||
|
||||
return api_key
|
||||
else:
|
||||
# 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)
|
||||
raise AuthError("Invalid token: API key not found", 403, service_id=service_id)
|
||||
|
||||
|
||||
def __get_token_issuer(auth_token):
|
||||
def _get_auth_token(req):
|
||||
auth_header = req.headers.get('Authorization', None)
|
||||
if not auth_header:
|
||||
raise AuthError('Unauthorized: authentication token must be provided', 401)
|
||||
|
||||
auth_scheme = auth_header[:7].title()
|
||||
|
||||
if auth_scheme != 'Bearer ':
|
||||
raise AuthError('Unauthorized: authentication bearer scheme must be used', 401)
|
||||
|
||||
return auth_header[7:]
|
||||
|
||||
|
||||
def _get_token_issuer(auth_token):
|
||||
try:
|
||||
issuer = get_token_issuer(auth_token)
|
||||
except TokenIssuerError:
|
||||
|
||||
@@ -84,9 +84,17 @@ class Config(object):
|
||||
# URL of api app (on AWS this is the internal api endpoint)
|
||||
API_HOST_NAME = os.getenv('API_HOST_NAME')
|
||||
|
||||
# secrets that internal apps, such as the admin app or document download, must use to authenticate with the API
|
||||
# LEGACY: replacing with INTERNAL_CLIENT_API_KEYS
|
||||
API_INTERNAL_SECRETS = json.loads(os.environ.get('API_INTERNAL_SECRETS', '[]'))
|
||||
|
||||
# secrets that internal apps, such as the admin app or document download, must use to authenticate with the API
|
||||
ADMIN_CLIENT_ID = 'notify-admin'
|
||||
GOVUK_ALERTS_CLIENT_ID = 'govuk-alerts'
|
||||
|
||||
INTERNAL_CLIENT_API_KEYS = {
|
||||
ADMIN_CLIENT_ID: API_INTERNAL_SECRETS
|
||||
}
|
||||
|
||||
# encyption secret/salt
|
||||
SECRET_KEY = os.getenv('SECRET_KEY')
|
||||
DANGEROUS_SALT = os.getenv('DANGEROUS_SALT')
|
||||
@@ -129,7 +137,6 @@ class Config(object):
|
||||
###########################
|
||||
|
||||
NOTIFY_ENVIRONMENT = 'development'
|
||||
ADMIN_CLIENT_USER_NAME = 'notify-admin'
|
||||
AWS_REGION = 'eu-west-1'
|
||||
INVITATION_EXPIRATION_DAYS = 2
|
||||
NOTIFY_APP_NAME = 'api'
|
||||
@@ -399,7 +406,11 @@ class Development(Config):
|
||||
TRANSIENT_UPLOADED_LETTERS = 'development-transient-uploaded-letters'
|
||||
LETTER_SANITISE_BUCKET_NAME = 'development-letters-sanitise'
|
||||
|
||||
API_INTERNAL_SECRETS = ['dev-notify-secret-key']
|
||||
INTERNAL_CLIENT_API_KEYS = {
|
||||
Config.ADMIN_CLIENT_ID: ['dev-notify-secret-key'],
|
||||
Config.GOVUK_ALERTS_CLIENT_ID: ['govuk-alerts-secret-key']
|
||||
}
|
||||
|
||||
SECRET_KEY = 'dev-notify-secret-key'
|
||||
DANGEROUS_SALT = 'dev-notify-salt'
|
||||
|
||||
|
||||
11
app/v2/govuk_alerts/__init__.py
Normal file
11
app/v2/govuk_alerts/__init__.py
Normal file
@@ -0,0 +1,11 @@
|
||||
from flask import Blueprint
|
||||
|
||||
from app.v2.errors import register_errors
|
||||
|
||||
v2_govuk_alerts_blueprint = Blueprint(
|
||||
"v2_govuk-alerts_blueprint",
|
||||
__name__,
|
||||
url_prefix='/v2/govuk-alerts',
|
||||
)
|
||||
|
||||
register_errors(v2_govuk_alerts_blueprint)
|
||||
8
app/v2/govuk_alerts/get_broadcasts.py
Normal file
8
app/v2/govuk_alerts/get_broadcasts.py
Normal file
@@ -0,0 +1,8 @@
|
||||
from flask import jsonify
|
||||
|
||||
from app.v2.govuk_alerts import v2_govuk_alerts_blueprint
|
||||
|
||||
|
||||
@v2_govuk_alerts_blueprint.route('')
|
||||
def get_broadcasts():
|
||||
return jsonify({})
|
||||
@@ -27,8 +27,8 @@ def create_authorization_header(service_id=None, key_type=KEY_TYPE_NORMAL):
|
||||
secret = api_key.secret
|
||||
|
||||
else:
|
||||
client_id = current_app.config['ADMIN_CLIENT_USER_NAME']
|
||||
secret = current_app.config['API_INTERNAL_SECRETS'][0]
|
||||
client_id = current_app.config['ADMIN_CLIENT_ID']
|
||||
secret = current_app.config['INTERNAL_CLIENT_API_KEYS'][client_id][0]
|
||||
|
||||
token = create_jwt_token(secret=secret, client_id=client_id)
|
||||
return 'Authorization', 'Bearer {}'.format(token)
|
||||
|
||||
@@ -1,478 +1,331 @@
|
||||
import time
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from unittest.mock import call
|
||||
|
||||
import jwt
|
||||
import pytest
|
||||
from flask import current_app, json, request
|
||||
from freezegun import freeze_time
|
||||
from flask import current_app, g, request
|
||||
from notifications_python_client.authentication import create_jwt_token
|
||||
|
||||
from app import api_user
|
||||
from app import db
|
||||
from app.authentication.auth import (
|
||||
GENERAL_TOKEN_ERROR_MESSAGE,
|
||||
AuthError,
|
||||
requires_admin_auth,
|
||||
_decode_jwt_token,
|
||||
_get_auth_token,
|
||||
_get_token_issuer,
|
||||
requires_auth,
|
||||
requires_internal_auth,
|
||||
)
|
||||
from app.dao.api_key_dao import (
|
||||
expire_api_key,
|
||||
get_model_api_keys,
|
||||
get_unsigned_secret,
|
||||
get_unsigned_secrets,
|
||||
save_model_api_key,
|
||||
)
|
||||
from app.dao.services_dao import dao_fetch_service_by_id
|
||||
from app.models import KEY_TYPE_NORMAL, ApiKey
|
||||
from tests.conftest import set_config, set_config_values
|
||||
from tests.conftest import set_config_values
|
||||
|
||||
|
||||
@pytest.mark.parametrize('auth_fn', [requires_auth, requires_admin_auth])
|
||||
def test_should_not_allow_request_with_no_token(client, auth_fn):
|
||||
@pytest.fixture
|
||||
def internal_jwt_token(notify_api):
|
||||
with set_config_values(notify_api, {
|
||||
'INTERNAL_CLIENT_API_KEYS': {
|
||||
'my-internal-app': ['my-internal-app-secret'],
|
||||
}
|
||||
}):
|
||||
yield create_jwt_token(
|
||||
client_id='my-internal-app',
|
||||
secret='my-internal-app-secret'
|
||||
)
|
||||
|
||||
|
||||
def requires_my_internal_app_auth():
|
||||
requires_internal_auth('my-internal-app')
|
||||
|
||||
|
||||
def create_custom_jwt_token(headers=None, payload=None, secret=None):
|
||||
# code copied from notifications_python_client.authentication.py::create_jwt_token
|
||||
headers = headers or {"typ": 'JWT', "alg": 'HS256'}
|
||||
return jwt.encode(payload=payload, key=secret or str(uuid.uuid4()), headers=headers)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def service_jwt_secret(sample_api_key):
|
||||
return get_unsigned_secrets(sample_api_key.service_id)[0]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def service_jwt_token(sample_api_key, service_jwt_secret):
|
||||
return create_jwt_token(
|
||||
secret=service_jwt_secret,
|
||||
client_id=str(sample_api_key.service_id),
|
||||
)
|
||||
|
||||
|
||||
def test_requires_auth_should_allow_valid_token_for_request(
|
||||
client,
|
||||
service_jwt_token,
|
||||
):
|
||||
response = client.get('/notifications', headers={'Authorization': 'Bearer {}'.format(service_jwt_token)})
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
def test_requires_admin_auth_should_allow_valid_token_for_request(client):
|
||||
admin_jwt_client_id = current_app.config['ADMIN_CLIENT_ID']
|
||||
admin_jwt_secret = current_app.config['INTERNAL_CLIENT_API_KEYS'][admin_jwt_client_id][0]
|
||||
admin_jwt_token = create_jwt_token(admin_jwt_secret, admin_jwt_client_id)
|
||||
|
||||
response = client.get('/service', headers={'Authorization': 'Bearer {}'.format(admin_jwt_token)})
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
def test_requires_govuk_alerts_auth_should_allow_valid_token_for_request(client):
|
||||
govuk_alerts_jwt_client_id = current_app.config['GOVUK_ALERTS_CLIENT_ID']
|
||||
govuk_alerts_jwt_secret = current_app.config['INTERNAL_CLIENT_API_KEYS'][govuk_alerts_jwt_client_id][0]
|
||||
govuk_alerts_jwt_token = create_jwt_token(govuk_alerts_jwt_secret, govuk_alerts_jwt_client_id)
|
||||
|
||||
response = client.get('/v2/govuk-alerts', headers={'Authorization': 'Bearer {}'.format(govuk_alerts_jwt_token)})
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
def test_get_auth_token_should_not_allow_request_with_no_token(client):
|
||||
request.headers = {}
|
||||
with pytest.raises(AuthError) as exc:
|
||||
auth_fn()
|
||||
_get_auth_token(request)
|
||||
assert exc.value.short_message == 'Unauthorized: authentication token must be provided'
|
||||
|
||||
|
||||
@pytest.mark.parametrize('auth_fn', [requires_auth, requires_admin_auth])
|
||||
def test_should_not_allow_request_with_incorrect_header(client, auth_fn):
|
||||
def test_get_auth_token_should_not_allow_request_with_incorrect_header(client):
|
||||
request.headers = {'Authorization': 'Basic 1234'}
|
||||
with pytest.raises(AuthError) as exc:
|
||||
auth_fn()
|
||||
_get_auth_token(request)
|
||||
assert exc.value.short_message == 'Unauthorized: authentication bearer scheme must be used'
|
||||
|
||||
|
||||
@pytest.mark.parametrize('auth_fn', [requires_auth, requires_admin_auth])
|
||||
def test_should_not_allow_request_with_incorrect_token(client, auth_fn):
|
||||
request.headers = {'Authorization': 'Bearer 1234'}
|
||||
@pytest.mark.parametrize('scheme', ['bearer', 'Bearer'])
|
||||
def test_get_auth_token_should_allow_valid_token(client, scheme):
|
||||
token = create_jwt_token(client_id='something', secret='secret')
|
||||
request.headers = {'Authorization': '{} {}'.format(scheme, token)}
|
||||
assert _get_auth_token(request) == token
|
||||
|
||||
|
||||
def test_get_token_issuer_should_not_allow_request_with_incorrect_token(client):
|
||||
with pytest.raises(AuthError) as exc:
|
||||
auth_fn()
|
||||
_get_token_issuer("Bearer 1234")
|
||||
assert exc.value.short_message == GENERAL_TOKEN_ERROR_MESSAGE
|
||||
|
||||
|
||||
@pytest.mark.parametrize('auth_fn', [requires_auth, requires_admin_auth])
|
||||
def test_should_not_allow_request_with_no_iss(client, auth_fn):
|
||||
# code copied from notifications_python_client.authentication.py::create_jwt_token
|
||||
headers = {
|
||||
"typ": 'JWT',
|
||||
"alg": 'HS256'
|
||||
}
|
||||
def test_get_token_issuer_should_not_allow_request_with_no_iss(client):
|
||||
token = create_custom_jwt_token(
|
||||
payload={'iat': int(time.time())}
|
||||
)
|
||||
|
||||
claims = {
|
||||
# 'iss': not provided
|
||||
'iat': int(time.time())
|
||||
}
|
||||
|
||||
token = jwt.encode(payload=claims, key=str(uuid.uuid4()), headers=headers)
|
||||
|
||||
request.headers = {'Authorization': 'Bearer {}'.format(token)}
|
||||
with pytest.raises(AuthError) as exc:
|
||||
auth_fn()
|
||||
_get_token_issuer(token)
|
||||
assert exc.value.short_message == 'Invalid token: iss field not provided'
|
||||
|
||||
|
||||
def test_auth_should_not_allow_request_with_no_iat(client, sample_api_key):
|
||||
iss = str(sample_api_key.service_id)
|
||||
# code copied from notifications_python_client.authentication.py::create_jwt_token
|
||||
headers = {
|
||||
"typ": 'JWT',
|
||||
"alg": 'HS256'
|
||||
}
|
||||
def test_decode_jwt_token_should_not_allow_non_hs256_algorithm(client, sample_api_key):
|
||||
token = create_custom_jwt_token(
|
||||
headers={"typ": 'JWT', "alg": 'HS512'},
|
||||
payload={},
|
||||
)
|
||||
|
||||
claims = {
|
||||
'iss': iss
|
||||
# 'iat': not provided
|
||||
}
|
||||
|
||||
token = jwt.encode(payload=claims, key=str(uuid.uuid4()), headers=headers)
|
||||
|
||||
request.headers = {'Authorization': 'Bearer {}'.format(token)}
|
||||
with pytest.raises(AuthError) as exc:
|
||||
requires_auth()
|
||||
assert exc.value.short_message == 'Invalid token: API key not found'
|
||||
|
||||
|
||||
def test_auth_should_not_allow_request_with_non_hs256_algorithm(client, sample_api_key):
|
||||
iss = str(sample_api_key.service_id)
|
||||
# code copied from notifications_python_client.authentication.py::create_jwt_token
|
||||
headers = {
|
||||
"typ": 'JWT',
|
||||
"alg": 'HS512'
|
||||
}
|
||||
|
||||
claims = {
|
||||
'iss': iss,
|
||||
'iat': int(time.time())
|
||||
}
|
||||
|
||||
token = jwt.encode(payload=claims, key=str(uuid.uuid4()), headers=headers)
|
||||
|
||||
request.headers = {'Authorization': 'Bearer {}'.format(token)}
|
||||
with pytest.raises(AuthError) as exc:
|
||||
requires_auth()
|
||||
_decode_jwt_token(token, [sample_api_key])
|
||||
assert exc.value.short_message == 'Invalid token: algorithm used is not HS256'
|
||||
|
||||
|
||||
def test_admin_auth_should_not_allow_request_with_no_iat(client):
|
||||
iss = current_app.config['ADMIN_CLIENT_USER_NAME']
|
||||
secret = current_app.config['API_INTERNAL_SECRETS'][0]
|
||||
def test_decode_jwt_token_should_not_allow_no_iat(
|
||||
client,
|
||||
sample_api_key,
|
||||
):
|
||||
token = create_custom_jwt_token(
|
||||
payload={'iss': 'something'}
|
||||
)
|
||||
|
||||
# code copied from notifications_python_client.authentication.py::create_jwt_token
|
||||
headers = {
|
||||
"typ": 'JWT',
|
||||
"alg": 'HS256'
|
||||
}
|
||||
|
||||
claims = {
|
||||
'iss': iss
|
||||
# 'iat': not provided
|
||||
}
|
||||
|
||||
token = jwt.encode(payload=claims, key=secret, headers=headers)
|
||||
|
||||
request.headers = {'Authorization': 'Bearer {}'.format(token)}
|
||||
with pytest.raises(AuthError) as exc:
|
||||
requires_admin_auth()
|
||||
assert exc.value.short_message == "Unauthorized: admin authentication token not found"
|
||||
_decode_jwt_token(token, [sample_api_key])
|
||||
assert exc.value.short_message == "Invalid token: API key not found"
|
||||
|
||||
|
||||
def test_admin_auth_should_not_allow_request_with_old_iat(client):
|
||||
iss = current_app.config['ADMIN_CLIENT_USER_NAME']
|
||||
secret = current_app.config['API_INTERNAL_SECRETS'][0]
|
||||
def test_decode_jwt_token_should_not_allow_old_iat(
|
||||
client,
|
||||
sample_api_key,
|
||||
):
|
||||
token = create_custom_jwt_token(
|
||||
payload={'iss': 'something', 'iat': int(time.time()) - 60},
|
||||
secret=sample_api_key.secret,
|
||||
)
|
||||
|
||||
# code copied from notifications_python_client.authentication.py::create_jwt_token
|
||||
headers = {
|
||||
"typ": 'JWT',
|
||||
"alg": 'HS256'
|
||||
}
|
||||
|
||||
claims = {
|
||||
'iss': iss,
|
||||
'iat': int(time.time()) - 60
|
||||
}
|
||||
|
||||
token = jwt.encode(payload=claims, key=secret, headers=headers)
|
||||
|
||||
request.headers = {'Authorization': 'Bearer {}'.format(token)}
|
||||
with pytest.raises(AuthError) as exc:
|
||||
requires_admin_auth()
|
||||
assert exc.value.short_message == "Invalid token: expired, check that your system clock is accurate"
|
||||
_decode_jwt_token(token, [sample_api_key])
|
||||
assert exc.value.short_message == "Error: Your system clock must be accurate to within 30 seconds"
|
||||
|
||||
|
||||
def test_auth_should_not_allow_request_with_extra_claims(client, sample_api_key):
|
||||
iss = str(sample_api_key.service_id)
|
||||
key = get_unsigned_secrets(sample_api_key.service_id)[0]
|
||||
def test_decode_jwt_token_should_not_allow_extra_claims(
|
||||
client,
|
||||
sample_api_key,
|
||||
):
|
||||
token = create_custom_jwt_token(
|
||||
payload={
|
||||
'iss': 'something',
|
||||
'iat': int(time.time()),
|
||||
'aud': 'notifications.service.gov.uk' # extra claim that we don't support
|
||||
},
|
||||
secret=sample_api_key.secret,
|
||||
)
|
||||
|
||||
headers = {
|
||||
"typ": 'JWT',
|
||||
"alg": 'HS256'
|
||||
}
|
||||
|
||||
claims = {
|
||||
'iss': iss,
|
||||
'iat': int(time.time()),
|
||||
'aud': 'notifications.service.gov.uk' # extra claim that we don't support
|
||||
}
|
||||
|
||||
token = jwt.encode(payload=claims, key=key, headers=headers)
|
||||
|
||||
request.headers = {'Authorization': 'Bearer {}'.format(token)}
|
||||
with pytest.raises(AuthError) as exc:
|
||||
requires_auth()
|
||||
_decode_jwt_token(token, [sample_api_key])
|
||||
assert exc.value.short_message == GENERAL_TOKEN_ERROR_MESSAGE
|
||||
|
||||
|
||||
def test_should_not_allow_invalid_secret(client, sample_api_key):
|
||||
def test_decode_jwt_token_should_not_allow_invalid_secret(
|
||||
client,
|
||||
sample_api_key
|
||||
):
|
||||
token = create_jwt_token(
|
||||
secret="not-so-secret",
|
||||
client_id=str(sample_api_key.service_id))
|
||||
response = client.get(
|
||||
'/notifications',
|
||||
headers={'Authorization': "Bearer {}".format(token)}
|
||||
client_id=str(sample_api_key.service_id)
|
||||
)
|
||||
assert response.status_code == 403
|
||||
data = json.loads(response.get_data())
|
||||
assert data['message'] == {"token": ['Invalid token: API key not found']}
|
||||
|
||||
with pytest.raises(AuthError) as exc:
|
||||
_decode_jwt_token(token, [sample_api_key])
|
||||
assert exc.value.short_message == 'Invalid token: API key not found'
|
||||
|
||||
|
||||
@pytest.mark.parametrize('scheme', ['bearer', 'Bearer'])
|
||||
def test_should_allow_valid_token(client, sample_api_key, scheme):
|
||||
token = __create_token(sample_api_key.service_id)
|
||||
response = client.get('/notifications', headers={'Authorization': '{} {}'.format(scheme, token)})
|
||||
assert response.status_code == 200
|
||||
def test_decode_jwt_token_should_allow_multiple_api_keys(
|
||||
client,
|
||||
sample_api_key,
|
||||
sample_test_api_key,
|
||||
):
|
||||
token = create_jwt_token(
|
||||
secret=sample_test_api_key.secret,
|
||||
client_id=str(sample_test_api_key.service_id),
|
||||
)
|
||||
|
||||
# successful if no error is raised
|
||||
_decode_jwt_token(token, [sample_api_key, sample_test_api_key])
|
||||
|
||||
|
||||
def test_decode_jwt_token_should_allow_some_expired_keys(
|
||||
client,
|
||||
sample_api_key,
|
||||
sample_test_api_key,
|
||||
):
|
||||
expire_api_key(sample_api_key.service_id, sample_api_key.id)
|
||||
|
||||
token = create_jwt_token(
|
||||
secret=sample_test_api_key.secret,
|
||||
client_id=str(sample_test_api_key.service_id),
|
||||
)
|
||||
|
||||
# successful if no error is raised
|
||||
_decode_jwt_token(token, [sample_api_key, sample_test_api_key])
|
||||
|
||||
|
||||
def test_decode_jwt_token_errors_when_all_api_keys_are_expired(
|
||||
client,
|
||||
sample_api_key,
|
||||
sample_test_api_key,
|
||||
):
|
||||
expire_api_key(sample_api_key.service_id, sample_api_key.id)
|
||||
expire_api_key(sample_test_api_key.service_id, sample_test_api_key.id)
|
||||
|
||||
token = create_jwt_token(
|
||||
secret=sample_test_api_key.secret,
|
||||
client_id=str(sample_test_api_key.service_id),
|
||||
)
|
||||
|
||||
with pytest.raises(AuthError) as exc:
|
||||
_decode_jwt_token(token, [sample_api_key, sample_test_api_key], service_id='1234')
|
||||
|
||||
assert exc.value.short_message == 'Invalid token: API key revoked'
|
||||
assert exc.value.service_id == '1234'
|
||||
assert exc.value.api_key_id == sample_test_api_key.id
|
||||
|
||||
|
||||
def test_decode_jwt_token_returns_error_with_no_secrets(client):
|
||||
with pytest.raises(AuthError) as exc:
|
||||
_decode_jwt_token('token', [])
|
||||
assert exc.value.short_message == "Invalid token: API key not found"
|
||||
|
||||
|
||||
@pytest.mark.parametrize('service_id', ['not-a-valid-id', 1234])
|
||||
def test_should_not_allow_service_id_that_is_not_the_wrong_data_type(client, sample_api_key, service_id):
|
||||
token = create_jwt_token(secret=get_unsigned_secrets(sample_api_key.service_id)[0],
|
||||
client_id=service_id)
|
||||
response = client.get(
|
||||
'/notifications',
|
||||
headers={'Authorization': "Bearer {}".format(token)}
|
||||
def test_requires_auth_should_not_allow_service_id_with_the_wrong_data_type(
|
||||
client,
|
||||
service_jwt_secret,
|
||||
service_id
|
||||
):
|
||||
token = create_jwt_token(
|
||||
client_id=service_id,
|
||||
secret=service_jwt_secret,
|
||||
)
|
||||
assert response.status_code == 403
|
||||
data = json.loads(response.get_data())
|
||||
assert data['message'] == {"token": ['Invalid token: service id is not the right data type']}
|
||||
|
||||
|
||||
def test_should_allow_valid_token_for_request_with_path_params_for_public_url(client, sample_api_key):
|
||||
token = __create_token(sample_api_key.service_id)
|
||||
response = client.get('/notifications', headers={'Authorization': 'Bearer {}'.format(token)})
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
def test_should_allow_valid_token_for_request_with_path_params_for_admin_url(client):
|
||||
token = create_jwt_token(
|
||||
current_app.config['API_INTERNAL_SECRETS'][0], current_app.config['ADMIN_CLIENT_USER_NAME']
|
||||
)
|
||||
response = client.get('/service', headers={'Authorization': 'Bearer {}'.format(token)})
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
def test_should_allow_valid_token_for_request_with_path_params_for_admin_url_with_second_secret(client):
|
||||
with set_config(client.application, 'API_INTERNAL_SECRETS', ["secret1", "secret2"]):
|
||||
token = create_jwt_token(
|
||||
current_app.config['API_INTERNAL_SECRETS'][0], current_app.config['ADMIN_CLIENT_USER_NAME']
|
||||
)
|
||||
response = client.get('/service', headers={'Authorization': 'Bearer {}'.format(token)})
|
||||
assert response.status_code == 200
|
||||
|
||||
token = create_jwt_token(
|
||||
current_app.config['API_INTERNAL_SECRETS'][1], current_app.config['ADMIN_CLIENT_USER_NAME']
|
||||
)
|
||||
response = client.get('/service', headers={'Authorization': 'Bearer {}'.format(token)})
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
def test_should_allow_valid_token_when_service_has_multiple_keys(client, sample_api_key):
|
||||
data = {'service': sample_api_key.service,
|
||||
'name': 'some key name',
|
||||
'created_by': sample_api_key.created_by,
|
||||
'key_type': KEY_TYPE_NORMAL
|
||||
}
|
||||
api_key = ApiKey(**data)
|
||||
save_model_api_key(api_key)
|
||||
token = __create_token(sample_api_key.service_id)
|
||||
response = client.get(
|
||||
'/notifications',
|
||||
headers={'Authorization': 'Bearer {}'.format(token)})
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
def test_authentication_passes_when_service_has_multiple_keys_some_expired(
|
||||
client,
|
||||
sample_api_key):
|
||||
expired_key_data = {'service': sample_api_key.service,
|
||||
'name': 'expired_key',
|
||||
'expiry_date': datetime.utcnow(),
|
||||
'created_by': sample_api_key.created_by,
|
||||
'key_type': KEY_TYPE_NORMAL
|
||||
}
|
||||
expired_key = ApiKey(**expired_key_data)
|
||||
save_model_api_key(expired_key)
|
||||
another_key = {'service': sample_api_key.service,
|
||||
'name': 'another_key',
|
||||
'created_by': sample_api_key.created_by,
|
||||
'key_type': KEY_TYPE_NORMAL
|
||||
}
|
||||
api_key = ApiKey(**another_key)
|
||||
save_model_api_key(api_key)
|
||||
token = create_jwt_token(
|
||||
secret=get_unsigned_secret(api_key.id),
|
||||
client_id=str(sample_api_key.service_id))
|
||||
response = client.get(
|
||||
'/notifications',
|
||||
headers={'Authorization': 'Bearer {}'.format(token)})
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
def test_authentication_returns_token_expired_when_service_uses_expired_key_and_has_multiple_keys(client,
|
||||
sample_api_key):
|
||||
expired_key = {'service': sample_api_key.service,
|
||||
'name': 'expired_key',
|
||||
'created_by': sample_api_key.created_by,
|
||||
'key_type': KEY_TYPE_NORMAL
|
||||
}
|
||||
expired_api_key = ApiKey(**expired_key)
|
||||
save_model_api_key(expired_api_key)
|
||||
another_key = {'service': sample_api_key.service,
|
||||
'name': 'another_key',
|
||||
'created_by': sample_api_key.created_by,
|
||||
'key_type': KEY_TYPE_NORMAL
|
||||
}
|
||||
api_key = ApiKey(**another_key)
|
||||
save_model_api_key(api_key)
|
||||
token = create_jwt_token(
|
||||
secret=get_unsigned_secret(expired_api_key.id),
|
||||
client_id=str(sample_api_key.service_id))
|
||||
expire_api_key(service_id=sample_api_key.service_id, api_key_id=expired_api_key.id)
|
||||
request.headers = {'Authorization': 'Bearer {}'.format(token)}
|
||||
request.headers = {'Authorization': "Bearer {}".format(token)}
|
||||
with pytest.raises(AuthError) as exc:
|
||||
requires_auth()
|
||||
assert exc.value.short_message == 'Invalid token: API key revoked'
|
||||
assert exc.value.service_id == str(expired_api_key.service_id)
|
||||
assert exc.value.api_key_id == expired_api_key.id
|
||||
assert exc.value.short_message == 'Invalid token: service id is not the right data type'
|
||||
|
||||
|
||||
def test_authentication_returns_error_when_admin_client_has_no_secrets(client):
|
||||
api_secret = current_app.config.get('API_INTERNAL_SECRETS')[0]
|
||||
api_service_id = current_app.config.get('ADMIN_CLIENT_USER_NAME')
|
||||
token = create_jwt_token(
|
||||
secret=api_secret,
|
||||
client_id=api_service_id
|
||||
)
|
||||
with set_config(client.application, 'API_INTERNAL_SECRETS', []):
|
||||
response = client.get(
|
||||
'/service',
|
||||
headers={'Authorization': 'Bearer {}'.format(token)})
|
||||
assert response.status_code == 401
|
||||
error_message = json.loads(response.get_data())
|
||||
assert error_message['message'] == {"token": ["Unauthorized: admin authentication token not found"]}
|
||||
|
||||
|
||||
def test_authentication_returns_error_when_admin_client_secret_is_invalid(client):
|
||||
api_secret = current_app.config.get('API_INTERNAL_SECRETS')[0]
|
||||
token = create_jwt_token(
|
||||
secret=api_secret,
|
||||
client_id=current_app.config.get('ADMIN_CLIENT_USER_NAME')
|
||||
)
|
||||
current_app.config['API_INTERNAL_SECRETS'][0] = 'something-wrong'
|
||||
response = client.get(
|
||||
'/service',
|
||||
headers={'Authorization': 'Bearer {}'.format(token)})
|
||||
assert response.status_code == 401
|
||||
error_message = json.loads(response.get_data())
|
||||
assert error_message['message'] == {"token": ["Unauthorized: admin authentication token not found"]}
|
||||
current_app.config['API_INTERNAL_SECRETS'][0] = api_secret
|
||||
|
||||
|
||||
def test_authentication_returns_error_when_service_doesnt_exit(
|
||||
def test_requires_auth_returns_error_when_service_doesnt_exist(
|
||||
client,
|
||||
sample_api_key
|
||||
):
|
||||
# get service ID and secret the wrong way around
|
||||
token = create_jwt_token(
|
||||
secret=str(sample_api_key.service_id),
|
||||
client_id=str(sample_api_key.id))
|
||||
|
||||
response = client.get(
|
||||
'/notifications',
|
||||
headers={'Authorization': 'Bearer {}'.format(token)}
|
||||
client_id=str(sample_api_key.id),
|
||||
)
|
||||
assert response.status_code == 403
|
||||
error_message = json.loads(response.get_data())
|
||||
assert error_message['message'] == {'token': ['Invalid token: service not found']}
|
||||
|
||||
|
||||
def test_authentication_returns_error_when_service_inactive(client, sample_api_key):
|
||||
sample_api_key.service.active = False
|
||||
token = create_jwt_token(secret=str(sample_api_key.id), client_id=str(sample_api_key.service_id))
|
||||
|
||||
response = client.get('/notifications', headers={'Authorization': 'Bearer {}'.format(token)})
|
||||
|
||||
assert response.status_code == 403
|
||||
error_message = json.loads(response.get_data())
|
||||
assert error_message['message'] == {'token': ['Invalid token: service is archived']}
|
||||
|
||||
|
||||
def test_authentication_returns_error_when_service_has_no_secrets(client,
|
||||
sample_service,
|
||||
fake_uuid):
|
||||
token = create_jwt_token(
|
||||
secret=fake_uuid,
|
||||
client_id=str(sample_service.id))
|
||||
|
||||
request.headers = {'Authorization': 'Bearer {}'.format(token)}
|
||||
with pytest.raises(AuthError) as exc:
|
||||
requires_auth()
|
||||
assert exc.value.short_message == 'Invalid token: service not found'
|
||||
|
||||
|
||||
def test_requires_auth_returns_error_when_service_inactive(
|
||||
client,
|
||||
sample_api_key,
|
||||
service_jwt_token,
|
||||
):
|
||||
sample_api_key.service.active = False
|
||||
|
||||
request.headers = {'Authorization': 'Bearer {}'.format(service_jwt_token)}
|
||||
with pytest.raises(AuthError) as exc:
|
||||
requires_auth()
|
||||
assert exc.value.short_message == 'Invalid token: service is archived'
|
||||
|
||||
|
||||
def test_requires_auth_should_assign_global_variables(
|
||||
client,
|
||||
sample_api_key,
|
||||
service_jwt_token,
|
||||
):
|
||||
request.headers = {'Authorization': 'Bearer {}'.format(service_jwt_token)}
|
||||
requires_auth()
|
||||
assert g.api_user.id == sample_api_key.id
|
||||
assert g.service_id == sample_api_key.service_id
|
||||
assert g.authenticated_service.id == str(sample_api_key.service_id)
|
||||
|
||||
|
||||
def test_requires_auth_errors_if_service_has_no_api_keys(
|
||||
client,
|
||||
sample_api_key,
|
||||
service_jwt_token,
|
||||
):
|
||||
db.session.delete(sample_api_key)
|
||||
request.headers = {'Authorization': 'Bearer {}'.format(service_jwt_token)}
|
||||
with pytest.raises(AuthError) as exc:
|
||||
requires_auth()
|
||||
assert exc.value.short_message == 'Invalid token: service has no API keys'
|
||||
assert exc.value.service_id == str(sample_service.id)
|
||||
|
||||
|
||||
def test_should_attach_the_current_api_key_to_current_app(notify_api, sample_service, sample_api_key):
|
||||
with notify_api.test_request_context(), notify_api.test_client() as client:
|
||||
token = __create_token(sample_api_key.service_id)
|
||||
response = client.get(
|
||||
'/notifications',
|
||||
headers={'Authorization': 'Bearer {}'.format(token)}
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert str(api_user.id) == str(sample_api_key.id)
|
||||
|
||||
|
||||
def test_should_return_403_when_token_is_expired(client,
|
||||
sample_api_key):
|
||||
with freeze_time('2001-01-01T12:00:00'):
|
||||
token = __create_token(sample_api_key.service_id)
|
||||
with freeze_time('2001-01-01T12:00:40'):
|
||||
with pytest.raises(AuthError) as exc:
|
||||
request.headers = {'Authorization': 'Bearer {}'.format(token)}
|
||||
requires_auth()
|
||||
assert exc.value.short_message == 'Error: Your system clock must be accurate to within 30 seconds'
|
||||
assert exc.value.service_id == str(sample_api_key.service_id)
|
||||
assert str(exc.value.api_key_id) == str(sample_api_key.id)
|
||||
|
||||
|
||||
def __create_token(service_id):
|
||||
return create_jwt_token(secret=get_unsigned_secrets(service_id)[0],
|
||||
client_id=str(service_id))
|
||||
|
||||
|
||||
@pytest.mark.parametrize('check_proxy_header,header_value,expected_status', [
|
||||
(True, 'key_1', 200),
|
||||
(True, 'wrong_key', 200),
|
||||
(False, 'key_1', 200),
|
||||
(False, 'wrong_key', 200),
|
||||
])
|
||||
def test_proxy_key_non_auth_endpoint(notify_api, check_proxy_header, header_value, expected_status):
|
||||
with set_config_values(notify_api, {
|
||||
'ROUTE_SECRET_KEY_1': 'key_1',
|
||||
'ROUTE_SECRET_KEY_2': '',
|
||||
'CHECK_PROXY_HEADER': check_proxy_header,
|
||||
}):
|
||||
|
||||
with notify_api.test_client() as client:
|
||||
response = client.get(
|
||||
path='/_status',
|
||||
headers=[
|
||||
('X-Custom-Forwarder', header_value),
|
||||
]
|
||||
)
|
||||
assert response.status_code == expected_status
|
||||
|
||||
|
||||
@pytest.mark.parametrize('check_proxy_header,header_value,expected_status', [
|
||||
(True, 'key_1', 200),
|
||||
(True, 'wrong_key', 403),
|
||||
(False, 'key_1', 200),
|
||||
(False, 'wrong_key', 200),
|
||||
])
|
||||
def test_proxy_key_on_admin_auth_endpoint(notify_api, check_proxy_header, header_value, expected_status):
|
||||
token = create_jwt_token(
|
||||
current_app.config['API_INTERNAL_SECRETS'][0], current_app.config['ADMIN_CLIENT_USER_NAME']
|
||||
)
|
||||
|
||||
with set_config_values(notify_api, {
|
||||
'ROUTE_SECRET_KEY_1': 'key_1',
|
||||
'ROUTE_SECRET_KEY_2': '',
|
||||
'CHECK_PROXY_HEADER': check_proxy_header,
|
||||
}):
|
||||
|
||||
with notify_api.test_client() as client:
|
||||
response = client.get(
|
||||
path='/service',
|
||||
headers=[
|
||||
('X-Custom-Forwarder', header_value),
|
||||
('Authorization', 'Bearer {}'.format(token))
|
||||
]
|
||||
)
|
||||
assert response.status_code == expected_status
|
||||
|
||||
|
||||
def test_should_cache_service_and_api_key_lookups(mocker, client, sample_api_key):
|
||||
|
||||
def test_requires_auth_should_cache_service_and_api_key_lookups(
|
||||
mocker,
|
||||
client,
|
||||
service_jwt_token
|
||||
):
|
||||
mock_get_api_keys = mocker.patch(
|
||||
'app.serialised_models.get_model_api_keys',
|
||||
wraps=get_model_api_keys,
|
||||
@@ -482,15 +335,49 @@ def test_should_cache_service_and_api_key_lookups(mocker, client, sample_api_key
|
||||
wraps=dao_fetch_service_by_id,
|
||||
)
|
||||
|
||||
for _ in range(5):
|
||||
token = __create_token(sample_api_key.service_id)
|
||||
client.get('/notifications', headers={
|
||||
'Authorization': f'Bearer {token}'
|
||||
})
|
||||
request.headers = {'Authorization': f'Bearer {service_jwt_token}'}
|
||||
requires_auth()
|
||||
requires_auth() # second request
|
||||
|
||||
assert mock_get_api_keys.call_args_list == [
|
||||
call(str(sample_api_key.service_id))
|
||||
]
|
||||
assert mock_get_service.call_args_list == [
|
||||
call(sample_api_key.service_id)
|
||||
]
|
||||
mock_get_api_keys.assert_called_once()
|
||||
mock_get_service.assert_called_once()
|
||||
|
||||
|
||||
def test_requires_internal_auth_checks_proxy_key(
|
||||
client,
|
||||
mocker,
|
||||
internal_jwt_token,
|
||||
):
|
||||
proxy_check_mock = mocker.patch(
|
||||
'app.authentication.auth.request_helper.check_proxy_header_before_request'
|
||||
)
|
||||
|
||||
request.headers = {'Authorization': 'Bearer {}'.format(internal_jwt_token)}
|
||||
requires_my_internal_app_auth()
|
||||
proxy_check_mock.assert_called_once()
|
||||
|
||||
|
||||
def test_requires_internal_auth_errors_for_unknown_app(client):
|
||||
with pytest.raises(TypeError) as exc:
|
||||
requires_internal_auth('another-app')
|
||||
assert str(exc.value) == 'Unknown client_id for internal auth'
|
||||
|
||||
|
||||
def test_requires_internal_auth_errors_for_api_app_mismatch(
|
||||
client,
|
||||
internal_jwt_token,
|
||||
service_jwt_token
|
||||
):
|
||||
request.headers = {'Authorization': 'Bearer {}'.format(service_jwt_token)}
|
||||
with pytest.raises(AuthError) as exc:
|
||||
requires_my_internal_app_auth()
|
||||
assert exc.value.short_message == 'Unauthorized: not allowed to perform this action'
|
||||
|
||||
|
||||
def test_requires_internal_auth_sets_global_variables(
|
||||
client,
|
||||
internal_jwt_token,
|
||||
):
|
||||
request.headers = {'Authorization': 'Bearer {}'.format(internal_jwt_token)}
|
||||
requires_my_internal_app_auth()
|
||||
assert g.service_id == 'my-internal-app'
|
||||
|
||||
Reference in New Issue
Block a user