mirror of
https://github.com/GSA/notifications-api.git
synced 2025-12-20 23:41:17 -05:00
add check for inactive services to auth handler
cleaned up some auth code to marginally improve efficiency of error checking and hopefully make it easier to read fixed some incorrect auth headers in the deactivate tests
This commit is contained in:
@@ -46,10 +46,19 @@ def requires_auth():
|
|||||||
return handle_admin_key(auth_token, current_app.config.get('ADMIN_CLIENT_SECRET'))
|
return handle_admin_key(auth_token, current_app.config.get('ADMIN_CLIENT_SECRET'))
|
||||||
|
|
||||||
try:
|
try:
|
||||||
api_keys = get_model_api_keys(client)
|
service = dao_fetch_service_by_id(client)
|
||||||
except DataError:
|
except DataError:
|
||||||
raise AuthError("Invalid token: service id is not the right data type", 403)
|
raise AuthError("Invalid token: service id is not the right data type", 403)
|
||||||
for api_key in api_keys:
|
except NoResultFound:
|
||||||
|
raise AuthError("Invalid token: service not found", 403)
|
||||||
|
|
||||||
|
if not service.api_keys:
|
||||||
|
raise AuthError("Invalid token: service has no API keys", 403)
|
||||||
|
|
||||||
|
if not service.active:
|
||||||
|
raise AuthError("Invalid token: service is archived", 403)
|
||||||
|
|
||||||
|
for api_key in service.api_keys:
|
||||||
try:
|
try:
|
||||||
get_decode_errors(auth_token, api_key.unsigned_secret)
|
get_decode_errors(auth_token, api_key.unsigned_secret)
|
||||||
except TokenDecodeError:
|
except TokenDecodeError:
|
||||||
@@ -60,15 +69,8 @@ def requires_auth():
|
|||||||
|
|
||||||
_request_ctx_stack.top.api_user = api_key
|
_request_ctx_stack.top.api_user = api_key
|
||||||
return
|
return
|
||||||
|
|
||||||
try:
|
|
||||||
dao_fetch_service_by_id(client)
|
|
||||||
except NoResultFound:
|
|
||||||
raise AuthError("Invalid token: service not found", 403)
|
|
||||||
|
|
||||||
if not api_keys:
|
|
||||||
raise AuthError("Invalid token: service has no API keys", 403)
|
|
||||||
else:
|
else:
|
||||||
|
# service has API keys, but none matching the one the user provided
|
||||||
raise AuthError("Invalid token: signature, api token is not valid", 403)
|
raise AuthError("Invalid token: signature, api token is not valid", 403)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -188,7 +188,7 @@ class ApiKey(db.Model, Versioned):
|
|||||||
name = db.Column(db.String(255), nullable=False)
|
name = db.Column(db.String(255), nullable=False)
|
||||||
secret = db.Column(db.String(255), unique=True, nullable=False)
|
secret = db.Column(db.String(255), unique=True, nullable=False)
|
||||||
service_id = db.Column(UUID(as_uuid=True), db.ForeignKey('services.id'), index=True, nullable=False)
|
service_id = db.Column(UUID(as_uuid=True), db.ForeignKey('services.id'), index=True, nullable=False)
|
||||||
service = db.relationship('Service', backref=db.backref('api_keys', lazy='dynamic'))
|
service = db.relationship('Service', backref='api_keys')
|
||||||
key_type = db.Column(db.String(255), db.ForeignKey('key_types.name'), index=True, nullable=False)
|
key_type = db.Column(db.String(255), db.ForeignKey('key_types.name'), index=True, nullable=False)
|
||||||
expiry_date = db.Column(db.DateTime)
|
expiry_date = db.Column(db.DateTime)
|
||||||
created_at = db.Column(
|
created_at = db.Column(
|
||||||
|
|||||||
@@ -223,6 +223,17 @@ def test_authentication_returns_error_when_service_doesnt_exit(
|
|||||||
assert error_message['message'] == {'token': ['Invalid token: service not found']}
|
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('/service', 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(notify_api,
|
def test_authentication_returns_error_when_service_has_no_secrets(notify_api,
|
||||||
sample_service,
|
sample_service,
|
||||||
fake_uuid):
|
fake_uuid):
|
||||||
|
|||||||
@@ -11,19 +11,19 @@ from tests.app.conftest import (
|
|||||||
|
|
||||||
|
|
||||||
def test_deactivate_only_allows_post(client, sample_service):
|
def test_deactivate_only_allows_post(client, sample_service):
|
||||||
auth_header = create_authorization_header(service_id=str(sample_service.id))
|
auth_header = create_authorization_header()
|
||||||
response = client.get('/service/{}/deactivate'.format(uuid.uuid4()), headers=[auth_header])
|
response = client.get('/service/{}/deactivate'.format(uuid.uuid4()), headers=[auth_header])
|
||||||
assert response.status_code == 405
|
assert response.status_code == 405
|
||||||
|
|
||||||
|
|
||||||
def test_deactivate_service_errors_with_bad_service_id(client, sample_service):
|
def test_deactivate_service_errors_with_bad_service_id(client, sample_service):
|
||||||
auth_header = create_authorization_header(service_id=str(sample_service.id))
|
auth_header = create_authorization_header()
|
||||||
response = client.post('/service/{}/deactivate'.format(uuid.uuid4()), headers=[auth_header])
|
response = client.post('/service/{}/deactivate'.format(uuid.uuid4()), headers=[auth_header])
|
||||||
assert response.status_code == 404
|
assert response.status_code == 404
|
||||||
|
|
||||||
|
|
||||||
def test_deactivating_inactive_service_does_nothing(client, sample_service):
|
def test_deactivating_inactive_service_does_nothing(client, sample_service):
|
||||||
auth_header = create_authorization_header(service_id=str(sample_service.id))
|
auth_header = create_authorization_header()
|
||||||
sample_service.active = False
|
sample_service.active = False
|
||||||
response = client.post('/service/{}/deactivate'.format(sample_service.id), headers=[auth_header])
|
response = client.post('/service/{}/deactivate'.format(sample_service.id), headers=[auth_header])
|
||||||
assert response.status_code == 204
|
assert response.status_code == 204
|
||||||
@@ -37,7 +37,7 @@ def deactivated_service(client, notify_db, notify_db_session, sample_service):
|
|||||||
create_api_key(notify_db, notify_db_session)
|
create_api_key(notify_db, notify_db_session)
|
||||||
create_api_key(notify_db, notify_db_session)
|
create_api_key(notify_db, notify_db_session)
|
||||||
|
|
||||||
auth_header = create_authorization_header(service_id=str(sample_service.id))
|
auth_header = create_authorization_header()
|
||||||
response = client.post('/service/{}/deactivate'.format(sample_service.id), headers=[auth_header])
|
response = client.post('/service/{}/deactivate'.format(sample_service.id), headers=[auth_header])
|
||||||
assert response.status_code == 204
|
assert response.status_code == 204
|
||||||
assert response.data == b''
|
assert response.data == b''
|
||||||
@@ -50,7 +50,7 @@ def test_deactivating_service_changes_name_and_email(deactivated_service):
|
|||||||
|
|
||||||
|
|
||||||
def test_deactivating_service_revokes_api_keys(deactivated_service):
|
def test_deactivating_service_revokes_api_keys(deactivated_service):
|
||||||
assert deactivated_service.api_keys.count() == 2
|
assert len(deactivated_service.api_keys) == 2
|
||||||
for key in deactivated_service.api_keys:
|
for key in deactivated_service.api_keys:
|
||||||
assert key.expiry_date is not None
|
assert key.expiry_date is not None
|
||||||
assert key.version == 2
|
assert key.version == 2
|
||||||
|
|||||||
Reference in New Issue
Block a user