mirror of
https://github.com/GSA/notifications-api.git
synced 2026-01-31 23:26:23 -05:00
Allow for multiple api keys for a service.
This commit is contained in:
@@ -1,13 +1,13 @@
|
||||
from flask import current_app
|
||||
from client.authentication import create_jwt_token
|
||||
|
||||
from app.dao.api_key_dao import get_unsigned_secret
|
||||
from app.dao.api_key_dao import get_unsigned_secrets
|
||||
|
||||
|
||||
def create_authorization_header(path, method, request_body=None, service_id=None):
|
||||
if service_id:
|
||||
client_id = service_id
|
||||
secret = get_unsigned_secret(service_id)
|
||||
secret = get_unsigned_secrets(service_id)[0]
|
||||
else:
|
||||
client_id = current_app.config.get('ADMIN_CLIENT_USER_NAME')
|
||||
secret = current_app.config.get('ADMIN_CLIENT_SECRET')
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
from client.authentication import create_jwt_token
|
||||
from flask import json, url_for
|
||||
|
||||
from app.dao.api_key_dao import get_unsigned_secret
|
||||
from app.dao.api_key_dao import get_unsigned_secrets, save_model_api_key
|
||||
from app.models import ApiKey
|
||||
|
||||
|
||||
def test_should_not_allow_request_with_no_token(notify_api):
|
||||
@@ -23,7 +24,7 @@ def test_should_not_allow_request_with_incorrect_header(notify_api):
|
||||
assert data['error'] == 'Unauthorized, authentication bearer scheme must be used'
|
||||
|
||||
|
||||
def test_should_not_allow_request_with_incorrect_token(notify_api):
|
||||
def test_should_not_allow_request_with_incorrect_token(notify_api, notify_db, notify_db_session, sample_api_key):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
response = client.get(url_for('service.get_service'),
|
||||
@@ -38,7 +39,7 @@ def test_should_not_allow_incorrect_path(notify_api, notify_db, notify_db_sessio
|
||||
with notify_api.test_client() as client:
|
||||
token = create_jwt_token(request_method="GET",
|
||||
request_path="/bad",
|
||||
secret=get_unsigned_secret(sample_api_key.service_id),
|
||||
secret=get_unsigned_secrets(sample_api_key.service_id)[0],
|
||||
client_id=sample_api_key.service_id)
|
||||
response = client.get(url_for('service.get_service'),
|
||||
headers={'Authorization': "Bearer {}".format(token)})
|
||||
@@ -79,6 +80,18 @@ def test_should_allow_valid_token(notify_api, notify_db, notify_db_session, samp
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
def test_should_allow_valid_token_when_service_has_multiple_keys(notify_api, notify_db, notify_db_session,
|
||||
sample_api_key):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
data = {'service_id': sample_api_key.service_id, 'name': 'some key name'}
|
||||
api_key = ApiKey(**data)
|
||||
save_model_api_key(api_key)
|
||||
token = __create_get_token(sample_api_key.service_id)
|
||||
response = client.get(url_for('service.get_service'),
|
||||
headers={'Authorization': 'Bearer {}'.format(token)})
|
||||
assert response.status_code == 200
|
||||
|
||||
JSON_BODY = json.dumps({
|
||||
'name': 'new name'
|
||||
})
|
||||
@@ -118,7 +131,7 @@ def test_should_not_allow_valid_token_with_invalid_post_body(notify_api, notify_
|
||||
def __create_get_token(service_id):
|
||||
return create_jwt_token(request_method="GET",
|
||||
request_path=url_for('service.get_service'),
|
||||
secret=get_unsigned_secret(service_id),
|
||||
secret=get_unsigned_secrets(service_id)[0],
|
||||
client_id=service_id)
|
||||
|
||||
|
||||
@@ -126,7 +139,7 @@ def __create_post_token(service_id, request_body):
|
||||
return create_jwt_token(
|
||||
request_method="POST",
|
||||
request_path=url_for('service.create_service'),
|
||||
secret=get_unsigned_secret(service_id),
|
||||
secret=get_unsigned_secrets(service_id)[0],
|
||||
client_id=service_id,
|
||||
request_body=request_body
|
||||
)
|
||||
|
||||
@@ -69,7 +69,7 @@ def sample_api_key(notify_db,
|
||||
service=None):
|
||||
if service is None:
|
||||
service = sample_service(notify_db, notify_db_session)
|
||||
data = {'service_id': service.id, 'name': service.name}
|
||||
data = {'service_id': service.id, 'name': uuid.uuid4()}
|
||||
api_key = ApiKey(**data)
|
||||
save_model_api_key(api_key)
|
||||
return api_key
|
||||
|
||||
@@ -5,6 +5,7 @@ from sqlalchemy.orm.exc import NoResultFound
|
||||
|
||||
from app.dao.api_key_dao import (save_model_api_key,
|
||||
get_model_api_keys,
|
||||
get_unsigned_secrets,
|
||||
get_unsigned_secret,
|
||||
_generate_secret,
|
||||
_get_secret)
|
||||
@@ -60,10 +61,20 @@ def test_should_return_api_key_for_service(notify_api, notify_db, notify_db_sess
|
||||
assert api_key == sample_api_key
|
||||
|
||||
|
||||
def test_should_return_unsigned_api_key_for_service_id(notify_api,
|
||||
notify_db,
|
||||
notify_db_session,
|
||||
sample_api_key):
|
||||
unsigned_api_key = get_unsigned_secret(sample_api_key.service_id)
|
||||
def test_should_return_unsigned_api_keys_for_service_id(notify_api,
|
||||
notify_db,
|
||||
notify_db_session,
|
||||
sample_api_key):
|
||||
unsigned_api_key = get_unsigned_secrets(sample_api_key.service_id)
|
||||
assert len(unsigned_api_key) == 1
|
||||
assert sample_api_key.secret != unsigned_api_key[0]
|
||||
assert unsigned_api_key[0] == _get_secret(sample_api_key.secret)
|
||||
|
||||
|
||||
def test_get_unsigned_secret_returns_key(notify_api,
|
||||
notify_db,
|
||||
notify_db_session,
|
||||
sample_api_key):
|
||||
unsigned_api_key = get_unsigned_secret(sample_api_key.id)
|
||||
assert sample_api_key.secret != unsigned_api_key
|
||||
assert unsigned_api_key == _get_secret(sample_api_key.secret)
|
||||
|
||||
Reference in New Issue
Block a user