darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-02-03 09:51:11 -05:00
Code Issues Packages Projects Releases Wiki Activity

Change Tokens to ApiKey

Browse Source 
Added name to ApiKey model
This commit is contained in:
Rebecca Law
2016-01-19 12:07:00 +00:00
parent 6966c2484f
commit 4fc5c34320
14 changed files with 281 additions and 226 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
app/authentication/auth.py
View File

@@ -2,7 +2,7 @@ from flask import request, jsonify, _request_ctx_stack
from client.authentication import decode_jwt_token, get_token_issuer from client.authentication import decode_jwt_token, get_token_issuer
from client.errors import TokenDecodeError, TokenRequestError, TokenExpiredError, TokenPayloadError from client.errors import TokenDecodeError, TokenRequestError, TokenExpiredError, TokenPayloadError
from app.dao.tokens_dao import get_unsigned_token from app.dao.api_key_dao import get_unsigned_secret
def authentication_response(message, code): def authentication_response(message, code):
@@ -49,5 +49,5 @@ def requires_auth():
def fetch_client(client): def fetch_client(client):
return { return {
"client": client, "client": client,
"secret": get_unsigned_token(client) "secret": get_unsigned_secret(client)
} }

51
app/dao/api_key_dao.py Normal file
View File

@@ -0,0 +1,51 @@
from flask import current_app
from itsdangerous import URLSafeSerializer
from app import db
from app.models import ApiKey
def save_model_api_key(api_key, update_dict={}):
if update_dict:
del update_dict['id']
db.session.query(ApiKey).filter_by(id=api_key.id).update(update_dict)
else:
api_key.secret = _generate_secret()
db.session.add(api_key)
db.session.commit()
def get_model_api_keys(service_id=None, raise_=True):
"""
:param raise_: when True query api_keys using one() which will raise NoResultFound exception
when False query api_keys usong first() which will return None and not raise an exception.
"""
if service_id:
# If expiry date is None the api_key is active
if raise_:
return ApiKey.query.filter_by(service_id=service_id, expiry_date=None).one()
else:
return ApiKey.query.filter_by(service_id=service_id, expiry_date=None).first()
return ApiKey.query.filter_by().all()
def get_unsigned_secret(service_id):
"""
There should only be one valid api_keys for each service.
This method can only be exposed to the Authentication of the api calls.
"""
api_key = ApiKey.query.filter_by(service_id=service_id, expiry_date=None).one()
return _get_secret(api_key.secret)
def _generate_secret(token=None):
import uuid
if not token:
token = uuid.uuid4()
serializer = URLSafeSerializer(current_app.config.get('SECRET_KEY'))
return serializer.dumps(str(token), current_app.config.get('DANGEROUS_SALT'))
def _get_secret(signed_secret):
serializer = URLSafeSerializer(current_app.config.get('SECRET_KEY'))
return serializer.loads(signed_secret, salt=current_app.config.get('DANGEROUS_SALT'))

51
app/dao/tokens_dao.py
View File

@@ -1,51 +0,0 @@
from flask import current_app
from itsdangerous import URLSafeSerializer
from app import db
from app.models import Token
def save_model_token(token, update_dict={}):
if update_dict:
del update_dict['id']
db.session.query(Token).filter_by(id=token.id).update(update_dict)
else:
token.token = _generate_token()
db.session.add(token)
db.session.commit()
def get_model_tokens(service_id=None, raise_=True):
"""
:param raise_: when True query tokens using one() which will raise NoResultFound exception
when False query tokens usong first() which will return None and not raise an exception.
"""
if service_id:
# If expiry date is None the token is active
if raise_:
return Token.query.filter_by(service_id=service_id, expiry_date=None).one()
else:
return Token.query.filter_by(service_id=service_id, expiry_date=None).first()
return Token.query.filter_by().all()
def get_unsigned_token(service_id):
"""
There should only be one valid token for each service.
This method can only be exposed to the Authentication of the api calls.
"""
token = Token.query.filter_by(service_id=service_id, expiry_date=None).one()
return _get_token(token.token)
def _generate_token(token=None):
import uuid
if not token:
token = uuid.uuid4()
serializer = URLSafeSerializer(current_app.config.get('SECRET_KEY'))
return serializer.dumps(str(token), current_app.config.get('DANGEROUS_SALT'))
def _get_token(token):
serializer = URLSafeSerializer(current_app.config.get('SECRET_KEY'))
return serializer.loads(token, salt=current_app.config.get('DANGEROUS_SALT'))

9
app/models.py
View File

@@ -85,13 +85,14 @@ class Service(db.Model):
restricted = db.Column(db.Boolean, index=False, unique=False, nullable=False) restricted = db.Column(db.Boolean, index=False, unique=False, nullable=False)
class Token(db.Model): class ApiKey(db.Model):
__tablename__ = 'tokens' __tablename__ = 'api_key'
id = db.Column(db.Integer, primary_key=True) id = db.Column(db.Integer, primary_key=True)
token = db.Column(db.String(255), unique=True, nullable=False) name = db.Column(db.String(255), nullable=False)
secret = db.Column(db.String(255), unique=True, nullable=False)
service_id = db.Column(db.Integer, db.ForeignKey('services.id'), index=True, nullable=False) service_id = db.Column(db.Integer, db.ForeignKey('services.id'), index=True, nullable=False)
service = db.relationship('Service', backref=db.backref('tokens', lazy='dynamic')) service = db.relationship('Service', backref=db.backref('api_key', lazy='dynamic'))
expiry_date = db.Column(db.DateTime) expiry_date = db.Column(db.DateTime)

11
app/schemas.py
View File

@@ -1,6 +1,5 @@
from . import ma from . import ma
from . import models from . import models
from marshmallow import post_load
# TODO I think marshmallow provides a better integration and error handling. # TODO I think marshmallow provides a better integration and error handling.
# Would be better to replace functionality in dao with the marshmallow supported # Would be better to replace functionality in dao with the marshmallow supported
@@ -19,7 +18,7 @@ class UserSchema(ma.ModelSchema):
class ServiceSchema(ma.ModelSchema): class ServiceSchema(ma.ModelSchema):
class Meta: class Meta:
model = models.Service model = models.Service
exclude = ("updated_at", "created_at", "tokens", "templates", "jobs") exclude = ("updated_at", "created_at", "api_key", "templates", "jobs")
class TemplateSchema(ma.ModelSchema): class TemplateSchema(ma.ModelSchema):
@@ -28,9 +27,9 @@ class TemplateSchema(ma.ModelSchema):
exclude = ("updated_at", "created_at", "service_id", "jobs") exclude = ("updated_at", "created_at", "service_id", "jobs")
class TokenSchema(ma.ModelSchema): class ApiKeySchema(ma.ModelSchema):
class Meta: class Meta:
model = models.Token model = models.ApiKey
exclude = ["service"] exclude = ["service"]
@@ -45,7 +44,7 @@ service_schema = ServiceSchema()
services_schema = ServiceSchema(many=True) services_schema = ServiceSchema(many=True)
template_schema = TemplateSchema() template_schema = TemplateSchema()
templates_schema = TemplateSchema(many=True) templates_schema = TemplateSchema(many=True)
token_schema = TokenSchema() api_key_schema = ApiKeySchema()
tokens_schema = TokenSchema(many=True) api_keys_schema = ApiKeySchema(many=True)
job_schema = JobSchema() job_schema = JobSchema()
jobs_schema = JobSchema(many=True) jobs_schema = JobSchema(many=True)

44
app/service/rest.py
View File

@@ -1,6 +1,6 @@
from datetime import datetime from datetime import datetime
from flask import (jsonify, request, current_app) from flask import (jsonify, request)
from sqlalchemy.exc import DataError from sqlalchemy.exc import DataError
from sqlalchemy.orm.exc import NoResultFound from sqlalchemy.orm.exc import NoResultFound
@@ -10,8 +10,8 @@ from app.dao.services_dao import (
save_model_service, get_model_services, delete_model_service) save_model_service, get_model_services, delete_model_service)
from app.dao.templates_dao import ( from app.dao.templates_dao import (
save_model_template, get_model_templates, delete_model_template) save_model_template, get_model_templates, delete_model_template)
from app.dao.tokens_dao import (save_model_token, get_model_tokens, get_unsigned_token) from app.dao.api_key_dao import (save_model_api_key, get_model_api_keys, get_unsigned_secret)
from app.models import Token from app.models import ApiKey
from app.schemas import ( from app.schemas import (
services_schema, service_schema, template_schema) services_schema, service_schema, template_schema)
@@ -29,13 +29,11 @@ def create_service():
# db.session.commit # db.session.commit
try: try:
save_model_service(service) save_model_service(service)
save_model_token(Token(service_id=service.id))
except DAOException as e: except DAOException as e:
return jsonify(result="error", message=str(e)), 400 return jsonify(result="error", message=str(e)), 400
return jsonify(data=service_schema.dump(service).data, token=get_unsigned_token(service.id)), 201 return jsonify(data=service_schema.dump(service).data), 201
# TODO auth to be added
@service.route('/<int:service_id>', methods=['PUT', 'DELETE']) @service.route('/<int:service_id>', methods=['PUT', 'DELETE'])
def update_service(service_id): def update_service(service_id):
try: try:
@@ -64,7 +62,6 @@ def update_service(service_id):
return jsonify(data=service_schema.dump(service).data), status_code return jsonify(data=service_schema.dump(service).data), status_code
# TODO auth to be added.
@service.route('/<int:service_id>', methods=['GET']) @service.route('/<int:service_id>', methods=['GET'])
@service.route('/', methods=['GET']) @service.route('/', methods=['GET'])
def get_service(service_id=None): def get_service(service_id=None):
@@ -78,9 +75,8 @@ def get_service(service_id=None):
return jsonify(data=data) return jsonify(data=data)
# TODO auth to be added @service.route('/<int:service_id>/api-key/renew', methods=['POST'])
@service.route('/<int:service_id>/token/renew', methods=['POST']) def renew_api_key(service_id=None):
def renew_token(service_id=None):
try: try:
get_model_services(service_id=service_id) get_model_services(service_id=service_id)
except DataError: except DataError:
@@ -89,20 +85,22 @@ def renew_token(service_id=None):
return jsonify(result="error", message="Service not found"), 404 return jsonify(result="error", message="Service not found"), 404
try: try:
service_token = get_model_tokens(service_id=service_id, raise_=False) service_api_key = get_model_api_keys(service_id=service_id, raise_=False)
if service_token: if service_api_key:
# expire existing token # expire existing api_key
save_model_token(service_token, update_dict={'id': service_token.id, 'expiry_date': datetime.now()}) save_model_api_key(service_api_key, update_dict={'id': service_api_key.id, 'expiry_date': datetime.now()})
# create a new one # create a new one
save_model_token(Token(service_id=service_id)) # TODO: what validation should be done here?
secret_name = request.get_json()['name']
save_model_api_key(ApiKey(service_id=service_id, name=secret_name))
except DAOException as e: except DAOException as e:
return jsonify(result='error', message=str(e)), 400 return jsonify(result='error', message=str(e)), 400
unsigned_token = get_unsigned_token(service_id) unsigned_api_key = get_unsigned_secret(service_id)
return jsonify(data=unsigned_token), 201 return jsonify(data=unsigned_api_key), 201
@service.route('/<int:service_id>/token/revoke', methods=['POST']) @service.route('/<int:service_id>/api-key/revoke', methods=['POST'])
def revoke_token(service_id): def revoke_api_key(service_id):
try: try:
get_model_services(service_id=service_id) get_model_services(service_id=service_id)
except DataError: except DataError:
@@ -110,13 +108,12 @@ def revoke_token(service_id):
except NoResultFound: except NoResultFound:
return jsonify(result="error", message="Service not found"), 404 return jsonify(result="error", message="Service not found"), 404
service_token = get_model_tokens(service_id=service_id, raise_=False) service_api_key = get_model_api_keys(service_id=service_id, raise_=False)
if service_token: if service_api_key:
save_model_token(service_token, update_dict={'id': service_token.id, 'expiry_date': datetime.now()}) save_model_api_key(service_api_key, update_dict={'id': service_api_key.id, 'expiry_date': datetime.now()})
return jsonify(), 202 return jsonify(), 202
# TODO auth to be added.
@service.route('/<int:service_id>/template/', methods=['POST']) @service.route('/<int:service_id>/template/', methods=['POST'])
def create_template(service_id): def create_template(service_id):
try: try:
@@ -135,7 +132,6 @@ def create_template(service_id):
return jsonify(data=template_schema.dump(template).data), 201 return jsonify(data=template_schema.dump(template).data), 201
# TODO auth to be added
@service.route('/<int:service_id>/template/<int:template_id>', methods=['PUT', 'DELETE']) @service.route('/<int:service_id>/template/<int:template_id>', methods=['PUT', 'DELETE'])
def update_template(service_id, template_id): def update_template(service_id, template_id):
try: try:

12
application.py
View File

@@ -26,10 +26,10 @@ def list_routes():
def create_admin_user_service(): def create_admin_user_service():
""" """
Convience method to create a admin user and service Convience method to create a admin user and service
:return: API token for admin service :return: API secret for admin service
""" """
from app.models import User, Service, Token from app.models import User, Service, ApiKey
from app.dao import tokens_dao, users_dao, services_dao from app.dao import api_key_dao, users_dao, services_dao
from flask import current_app from flask import current_app
user = User(**{'email_address': current_app.config['ADMIN_USER_EMAIL_ADDRESS']}) user = User(**{'email_address': current_app.config['ADMIN_USER_EMAIL_ADDRESS']})
@@ -41,9 +41,9 @@ def create_admin_user_service():
'active': True, 'active': True,
'restricted': True}) 'restricted': True})
services_dao.save_model_service(service) services_dao.save_model_service(service)
token = Token(**{'service_id': service.id}) api_key = ApiKey(**{'service_id': service.id})
tokens_dao.save_model_token(token) api_key_dao.save_model_api_key(api_key)
print('Token: {}'.format(tokens_dao.get_unsigned_token(service.id))) print('ApiKey: {}'.format(api_key_dao.get_unsigned_secret(service.id)))
if __name__ == '__main__': if __name__ == '__main__':

47
migrations/versions/0007_change_to_api_keys.py Normal file
View File

@@ -0,0 +1,47 @@
"""empty message
Revision ID: 0007_change_to_api_keys
Revises: 0005_add_job_details
Create Date: 2016-01-19 10:50:46.269618
"""
# revision identifiers, used by Alembic.
revision = '0007_change_to_api_keys'
down_revision = '0005_add_job_details'
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
def upgrade():
### commands auto generated by Alembic - please adjust! ###
op.create_table('api_key',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('name', sa.String(length=255), nullable=False),
sa.Column('secret', sa.String(length=255), nullable=False),
sa.Column('service_id', sa.Integer(), nullable=False),
sa.Column('expiry_date', sa.DateTime(), nullable=True),
sa.ForeignKeyConstraint(['service_id'], ['services.id'], ),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('secret')
)
op.create_index(op.f('ix_api_key_service_id'), 'api_key', ['service_id'], unique=False)
op.drop_table('tokens')
### end Alembic commands ###
def downgrade():
### commands auto generated by Alembic - please adjust! ###
op.create_table('tokens',
sa.Column('id', sa.INTEGER(), nullable=False),
sa.Column('token', sa.VARCHAR(length=255), autoincrement=False, nullable=False),
sa.Column('service_id', sa.INTEGER(), autoincrement=False, nullable=False),
sa.Column('expiry_date', postgresql.TIMESTAMP(), autoincrement=False, nullable=True),
sa.ForeignKeyConstraint(['service_id'], ['services.id'], name='tokens_service_id_fkey'),
sa.PrimaryKeyConstraint('id', name='tokens_pkey'),
sa.UniqueConstraint('token', name='tokens_token_key')
)
op.drop_index(op.f('ix_api_key_service_id'), table_name='api_key')
op.drop_table('api_key')
### end Alembic commands ###

6
tests/__init__.py
View File

@@ -1,6 +1,6 @@
from client.authentication import create_jwt_token from client.authentication import create_jwt_token
from app.dao.tokens_dao import get_unsigned_token from app.dao.api_key_dao import get_unsigned_secret
def create_authorization_header(service_id, path, method, request_body=None): def create_authorization_header(service_id, path, method, request_body=None):
@@ -8,14 +8,14 @@ def create_authorization_header(service_id, path, method, request_body=None):
token = create_jwt_token( token = create_jwt_token(
request_method=method, request_method=method,
request_path=path, request_path=path,
secret=get_unsigned_token(service_id), secret=get_unsigned_secret(service_id),
client_id=service_id, client_id=service_id,
request_body=request_body) request_body=request_body)
else: else:
token = create_jwt_token(request_method=method, token = create_jwt_token(request_method=method,
request_path=path, request_path=path,
secret=get_unsigned_token(service_id), secret=get_unsigned_secret(service_id),
client_id=service_id) client_id=service_id)
return 'Authorization', 'Bearer {}'.format(token) return 'Authorization', 'Bearer {}'.format(token)

32
tests/app/authentication/test_authentication.py
View File

@@ -1,7 +1,7 @@
from client.authentication import create_jwt_token from client.authentication import create_jwt_token
from flask import json, url_for from flask import json, url_for
from app.dao.tokens_dao import get_unsigned_token from app.dao.api_key_dao import get_unsigned_secret
def test_should_not_allow_request_with_no_token(notify_api): def test_should_not_allow_request_with_no_token(notify_api):
@@ -33,13 +33,13 @@ def test_should_not_allow_request_with_incorrect_token(notify_api):
assert data['error'] == 'Invalid token: signature' assert data['error'] == 'Invalid token: signature'
def test_should_not_allow_incorrect_path(notify_api, notify_db, notify_db_session, sample_token): def test_should_not_allow_incorrect_path(notify_api, notify_db, notify_db_session, sample_api_key):
with notify_api.test_request_context(): with notify_api.test_request_context():
with notify_api.test_client() as client: with notify_api.test_client() as client:
token = create_jwt_token(request_method="GET", token = create_jwt_token(request_method="GET",
request_path="/bad", request_path="/bad",
secret=get_unsigned_token(sample_token.service_id), secret=get_unsigned_secret(sample_api_key.service_id),
client_id=sample_token.service_id) client_id=sample_api_key.service_id)
response = client.get(url_for('status.show_status'), response = client.get(url_for('status.show_status'),
headers={'Authorization': "Bearer {}".format(token)}) headers={'Authorization': "Bearer {}".format(token)})
assert response.status_code == 403 assert response.status_code == 403
@@ -47,10 +47,10 @@ def test_should_not_allow_incorrect_path(notify_api, notify_db, notify_db_sessio
assert data['error'] == 'Invalid token: request' assert data['error'] == 'Invalid token: request'
def test_should_not_allow_incorrect_method(notify_api, notify_db, notify_db_session, sample_token): def test_should_not_allow_incorrect_method(notify_api, notify_db, notify_db_session, sample_api_key):
with notify_api.test_request_context(): with notify_api.test_request_context():
with notify_api.test_client() as client: with notify_api.test_client() as client:
token = __create_post_token(sample_token.service_id, {}) token = __create_post_token(sample_api_key.service_id, {})
response = client.get(url_for('status.show_status'), response = client.get(url_for('status.show_status'),
headers={'Authorization': "Bearer {}".format(token)}) headers={'Authorization': "Bearer {}".format(token)})
assert response.status_code == 403 assert response.status_code == 403
@@ -58,11 +58,11 @@ def test_should_not_allow_incorrect_method(notify_api, notify_db, notify_db_sess
assert data['error'] == 'Invalid token: request' assert data['error'] == 'Invalid token: request'
def test_should_not_allow_invalid_secret(notify_api, notify_db, notify_db_session, sample_token): def test_should_not_allow_invalid_secret(notify_api, notify_db, notify_db_session, sample_api_key):
with notify_api.test_request_context(): with notify_api.test_request_context():
with notify_api.test_client() as client: with notify_api.test_client() as client:
token = create_jwt_token(request_method="POST", request_path="/", secret="not-so-secret", token = create_jwt_token(request_method="POST", request_path="/", secret="not-so-secret",
client_id=sample_token.service_id) client_id=sample_api_key.service_id)
response = client.get(url_for('status.show_status'), response = client.get(url_for('status.show_status'),
headers={'Authorization': "Bearer {}".format(token)}) headers={'Authorization': "Bearer {}".format(token)})
assert response.status_code == 403 assert response.status_code == 403
@@ -70,10 +70,10 @@ def test_should_not_allow_invalid_secret(notify_api, notify_db, notify_db_sessio
assert data['error'] == 'Invalid token: signature' assert data['error'] == 'Invalid token: signature'
def test_should_allow_valid_token(notify_api, notify_db, notify_db_session, sample_token): def test_should_allow_valid_token(notify_api, notify_db, notify_db_session, sample_api_key):
with notify_api.test_request_context(): with notify_api.test_request_context():
with notify_api.test_client() as client: with notify_api.test_client() as client:
token = __create_get_token(sample_token.service_id) token = __create_get_token(sample_api_key.service_id)
response = client.get(url_for('status.show_status'), response = client.get(url_for('status.show_status'),
headers={'Authorization': 'Bearer {}'.format(token)}) headers={'Authorization': 'Bearer {}'.format(token)})
assert response.status_code == 200 assert response.status_code == 200
@@ -86,20 +86,20 @@ JSON_BODY = json.dumps({
}) })
def test_should_allow_valid_token_with_post_body(notify_api, notify_db, notify_db_session, sample_token): def test_should_allow_valid_token_with_post_body(notify_api, notify_db, notify_db_session, sample_api_key):
with notify_api.test_request_context(): with notify_api.test_request_context():
with notify_api.test_client() as client: with notify_api.test_client() as client:
token = __create_post_token(sample_token.service_id, JSON_BODY) token = __create_post_token(sample_api_key.service_id, JSON_BODY)
response = client.post(url_for('status.show_status'), response = client.post(url_for('status.show_status'),
data=JSON_BODY, data=JSON_BODY,
headers={'Authorization': 'Bearer {}'.format(token)}) headers={'Authorization': 'Bearer {}'.format(token)})
assert response.status_code == 200 assert response.status_code == 200
def test_should_not_allow_valid_token_with_invalid_post_body(notify_api, notify_db, notify_db_session, sample_token): def test_should_not_allow_valid_token_with_invalid_post_body(notify_api, notify_db, notify_db_session, sample_api_key):
with notify_api.test_request_context(): with notify_api.test_request_context():
with notify_api.test_client() as client: with notify_api.test_client() as client:
token = __create_post_token(sample_token.service_id, JSON_BODY) token = __create_post_token(sample_api_key.service_id, JSON_BODY)
response = client.post(url_for('status.show_status'), response = client.post(url_for('status.show_status'),
data="spurious", data="spurious",
headers={'Authorization': 'Bearer {}'.format(token)}) headers={'Authorization': 'Bearer {}'.format(token)})
@@ -111,7 +111,7 @@ def test_should_not_allow_valid_token_with_invalid_post_body(notify_api, notify_
def __create_get_token(service_id): def __create_get_token(service_id):
return create_jwt_token(request_method="GET", return create_jwt_token(request_method="GET",
request_path=url_for('status.show_status'), request_path=url_for('status.show_status'),
secret=get_unsigned_token(service_id), secret=get_unsigned_secret(service_id),
client_id=service_id) client_id=service_id)
@@ -119,7 +119,7 @@ def __create_post_token(service_id, request_body):
return create_jwt_token( return create_jwt_token(
request_method="POST", request_method="POST",
request_path=url_for('status.show_status'), request_path=url_for('status.show_status'),
secret=get_unsigned_token(service_id), secret=get_unsigned_secret(service_id),
client_id=service_id, client_id=service_id,
request_body=request_body request_body=request_body
) )

26
tests/app/conftest.py
View File

@@ -1,9 +1,9 @@
import pytest import pytest
from app.models import (User, Service, Template, Token, Job) from app.models import (User, Service, Template, ApiKey, Job)
from app.dao.users_dao import (save_model_user) from app.dao.users_dao import (save_model_user)
from app.dao.services_dao import save_model_service from app.dao.services_dao import save_model_service
from app.dao.templates_dao import save_model_template from app.dao.templates_dao import save_model_template
from app.dao.tokens_dao import save_model_token from app.dao.api_key_dao import save_model_api_key
from app.dao.jobs_dao import save_job from app.dao.jobs_dao import save_job
import uuid import uuid
@@ -51,7 +51,7 @@ def sample_template(notify_db,
service=None): service=None):
if service is None: if service is None:
service = sample_service(notify_db, notify_db_session) service = sample_service(notify_db, notify_db_session)
sample_token(notify_db, notify_db_session, service=service) sample_api_key(notify_db, notify_db_session, service=service)
data = { data = {
'name': template_name, 'name': template_name,
'template_type': template_type, 'template_type': template_type,
@@ -64,15 +64,15 @@ def sample_template(notify_db,
@pytest.fixture(scope='function') @pytest.fixture(scope='function')
def sample_token(notify_db, def sample_api_key(notify_db,
notify_db_session, notify_db_session,
service=None): service=None):
if service is None: if service is None:
service = sample_service(notify_db, notify_db_session) service = sample_service(notify_db, notify_db_session)
data = {'service_id': service.id} data = {'service_id': service.id, 'name': service.name}
token = Token(**data) api_key = ApiKey(**data)
save_model_token(token) save_model_api_key(api_key)
return token return api_key
@pytest.fixture(scope='function') @pytest.fixture(scope='function')
@@ -105,7 +105,7 @@ def sample_job(notify_db,
def sample_admin_service_id(notify_db, notify_db_session): def sample_admin_service_id(notify_db, notify_db_session):
admin_user = sample_user(notify_db, notify_db_session, email="notify_admin@digital.cabinet-office.gov.uk") admin_user = sample_user(notify_db, notify_db_session, email="notify_admin@digital.cabinet-office.gov.uk")
admin_service = sample_service(notify_db, notify_db_session, service_name="Sample Admin Service", user=admin_user) admin_service = sample_service(notify_db, notify_db_session, service_name="Sample Admin Service", user=admin_user)
data = {'service_id': admin_service.id} data = {'service_id': admin_service.id, 'name': 'sample admin key'}
token = Token(**data) api_key = ApiKey(**data)
save_model_token(token) save_model_api_key(api_key)
return admin_service.id return admin_service.id

69
tests/app/dao/test_api_key_dao.py Normal file
View File

@@ -0,0 +1,69 @@
from datetime import datetime
from pytest import fail
from sqlalchemy.orm.exc import NoResultFound
from app.dao.api_key_dao import (save_model_api_key,
get_model_api_keys,
get_unsigned_secret,
_generate_secret,
_get_secret)
from app.models import ApiKey
def test_secret_is_signed_and_can_be_read_again(notify_api):
import uuid
with notify_api.test_request_context():
token = str(uuid.uuid4())
signed_secret = _generate_secret(token=token)
assert token == _get_secret(signed_secret)
assert signed_secret != token
def test_save_api_key_should_create_new_api_key(notify_api, notify_db, notify_db_session, sample_service):
api_key = ApiKey(**{'service_id': sample_service.id, 'name': sample_service.name})
save_model_api_key(api_key)
all_api_keys = get_model_api_keys()
assert len(all_api_keys) == 1
assert all_api_keys[0] == api_key
def test_save_api_key_should_update_the_api_key(notify_api, notify_db, notify_db_session, sample_api_key):
now = datetime.utcnow()
saved_api_key = get_model_api_keys(sample_api_key.service_id)
save_model_api_key(saved_api_key, update_dict={'id': saved_api_key.id, 'expiry_date': now})
all_api_keys = get_model_api_keys()
assert len(all_api_keys) == 1
assert all_api_keys[0].expiry_date == now
assert all_api_keys[0].secret == saved_api_key.secret
assert all_api_keys[0].id == saved_api_key.id
assert all_api_keys[0].service_id == saved_api_key.service_id
def test_get_api_key_should_raise_exception_when_service_does_not_exist(notify_api, notify_db, notify_db_session,
sample_service):
try:
get_model_api_keys(sample_service.id)
fail("Should have thrown a NoResultFound exception")
except NoResultFound:
pass
def test_get_api_key_should_return_none_when_service_does_not_exist(notify_api, notify_db, notify_db_session,
sample_service):
assert get_model_api_keys(service_id=sample_service.id, raise_=False) is None
def test_should_return_api_key_for_service(notify_api, notify_db, notify_db_session, sample_api_key):
api_key = get_model_api_keys(sample_api_key.service_id)
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)
assert sample_api_key.secret != unsigned_api_key
assert unsigned_api_key == _get_secret(sample_api_key.secret)

62
tests/app/dao/test_tokens_dao.py
View File

@@ -1,62 +0,0 @@
import uuid
from app.dao.tokens_dao import (save_model_token, get_model_tokens, get_unsigned_token, _generate_token, _get_token)
from datetime import datetime
from app.models import Token
from pytest import fail
from sqlalchemy.orm.exc import NoResultFound
def test_token_is_signed_and_can_be_read_again(notify_api):
import uuid
with notify_api.test_request_context():
token = str(uuid.uuid4())
signed_token = _generate_token(token=token)
assert token == _get_token(signed_token)
assert signed_token != token
def test_save_token_should_create_new_token(notify_api, notify_db, notify_db_session, sample_service):
api_token = Token(**{'service_id': sample_service.id})
save_model_token(api_token)
all_tokens = get_model_tokens()
assert len(all_tokens) == 1
assert all_tokens[0] == api_token
def test_save_token_should_update_the_token(notify_api, notify_db, notify_db_session, sample_token):
now = datetime.utcnow()
saved_token = get_model_tokens(sample_token.service_id)
save_model_token(saved_token, update_dict={'id': saved_token.id, 'expiry_date': now})
all_tokens = get_model_tokens()
assert len(all_tokens) == 1
assert all_tokens[0].expiry_date == now
assert all_tokens[0].token == saved_token.token
assert all_tokens[0].id == saved_token.id
assert all_tokens[0].service_id == saved_token.service_id
def test_get_token_should_raise_exception_when_service_does_not_exist(notify_api, notify_db, notify_db_session,
sample_service):
try:
get_model_tokens(sample_service.id)
fail("Should have thrown a NoResultFound exception")
except NoResultFound:
pass
def test_get_token_should_return_none_when_service_does_not_exist(notify_api, notify_db, notify_db_session,
sample_service):
assert get_model_tokens(service_id=sample_service.id, raise_=False) is None
def test_should_return_token_for_service(notify_api, notify_db, notify_db_session, sample_token):
token = get_model_tokens(sample_token.service_id)
assert token == sample_token
def test_should_return_unsigned_token_for_service_id(notify_api, notify_db, notify_db_session,
sample_token):
unsigned_token = get_unsigned_token(sample_token.service_id)
assert sample_token.token != unsigned_token
assert unsigned_token == _get_token(sample_token.token)

83
tests/app/service/test_rest.py
View File

@@ -1,7 +1,7 @@
import json import json
from flask import url_for from flask import url_for
from app.dao.services_dao import save_model_service from app.dao.services_dao import save_model_service
from app.models import (Service, Token, Template) from app.models import (Service, ApiKey, Template)
from tests import create_authorization_header from tests import create_authorization_header
from tests.app.conftest import sample_user as create_sample_user from tests.app.conftest import sample_user as create_sample_user
@@ -70,7 +70,6 @@ def test_post_service(notify_api, notify_db, notify_db_session, sample_user, sam
json_resp = json.loads(resp.get_data(as_text=True)) json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['data']['name'] == service.name assert json_resp['data']['name'] == service.name
assert json_resp['data']['limit'] == service.limit assert json_resp['data']['limit'] == service.limit
assert json_resp['token'] is not None
def test_post_service_multiple_users(notify_api, notify_db, notify_db_session, sample_user, sample_admin_service_id): def test_post_service_multiple_users(notify_api, notify_db, notify_db_session, sample_user, sample_admin_service_id):
@@ -311,76 +310,84 @@ def test_delete_service_not_exists(notify_api, notify_db, notify_db_session, sam
assert Service.query.count() == 2 assert Service.query.count() == 2
def test_renew_token_should_return_token_when_service_does_not_have_a_valid_token(notify_api, notify_db, def test_renew_api_key_should_create_new_api_key_for_service(notify_api, notify_db,
notify_db_session, notify_db_session,
sample_service, sample_service,
sample_admin_service_id): sample_admin_service_id):
with notify_api.test_request_context(): with notify_api.test_request_context():
with notify_api.test_client() as client: with notify_api.test_client() as client:
data = {'name': 'some secret name'}
auth_header = create_authorization_header(service_id=sample_admin_service_id, auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.renew_token', service_id=sample_service.id), path=url_for('service.renew_api_key',
method='POST') service_id=sample_service.id),
response = client.post(url_for('service.renew_token', service_id=sample_service.id), method='POST',
request_body=json.dumps(data))
response = client.post(url_for('service.renew_api_key', service_id=sample_service.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header]) headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 201 assert response.status_code == 201
assert response.get_data is not None assert response.get_data is not None
saved_token = Token.query.filter_by(service_id=sample_service.id).first() saved_api_key = ApiKey.query.filter_by(service_id=sample_service.id).first()
assert saved_token.service_id == sample_service.id assert saved_api_key.service_id == sample_service.id
assert saved_api_key.name == 'some secret name'
def test_renew_token_should_expire_the_old_token_and_create_a_new_token(notify_api, notify_db, notify_db_session, def test_renew_api_key_should_expire_the_old_api_key_and_create_a_new_api_key(notify_api, notify_db, notify_db_session,
sample_token, sample_admin_service_id): sample_api_key, sample_admin_service_id):
with notify_api.test_request_context(): with notify_api.test_request_context():
with notify_api.test_client() as client: with notify_api.test_client() as client:
assert Token.query.count() == 2 assert ApiKey.query.count() == 2
data = {'name': 'some secret name'}
auth_header = create_authorization_header(service_id=sample_admin_service_id, auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.renew_token', path=url_for('service.renew_api_key',
service_id=sample_token.service_id), service_id=sample_api_key.service_id),
method='POST') method='POST',
request_body=json.dumps(data))
response = client.post(url_for('service.renew_token', service_id=sample_token.service_id), response = client.post(url_for('service.renew_api_key', service_id=sample_api_key.service_id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header]) headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 201 assert response.status_code == 201
assert Token.query.count() == 3 assert ApiKey.query.count() == 3
all_tokens = Token.query.filter_by(service_id=sample_token.service_id).all() all_api_keys = ApiKey.query.filter_by(service_id=sample_api_key.service_id).all()
for x in all_tokens: for x in all_api_keys:
if x.id == sample_token.id: if x.id == sample_api_key.id:
assert x.expiry_date is not None assert x.expiry_date is not None
else: else:
assert x.expiry_date is None assert x.expiry_date is None
assert x.token is not sample_token.token assert x.secret is not sample_api_key.secret
def test_create_token_should_return_error_when_service_does_not_exist(notify_api, notify_db, notify_db_session, def test_renew_api_key_should_return_error_when_service_does_not_exist(notify_api, notify_db, notify_db_session,
sample_service, sample_admin_service_id): sample_service, sample_admin_service_id):
with notify_api.test_request_context(): with notify_api.test_request_context():
with notify_api.test_client() as client: with notify_api.test_client() as client:
auth_header = create_authorization_header(service_id=sample_admin_service_id, auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.renew_token', service_id="123"), path=url_for('service.renew_api_key', service_id="123"),
method='POST') method='POST')
response = client.post(url_for('service.renew_token', service_id=123), response = client.post(url_for('service.renew_api_key', service_id=123),
headers=[('Content-Type', 'application/json'), auth_header]) headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 404 assert response.status_code == 404
def test_revoke_token_should_expire_token_for_service(notify_api, notify_db, notify_db_session, def test_revoke_api_key_should_expire_api_key_for_service(notify_api, notify_db, notify_db_session,
sample_token, sample_admin_service_id): sample_api_key, sample_admin_service_id):
with notify_api.test_request_context(): with notify_api.test_request_context():
with notify_api.test_client() as client: with notify_api.test_client() as client:
assert Token.query.count() == 2 assert ApiKey.query.count() == 2
auth_header = create_authorization_header(service_id=sample_admin_service_id, auth_header = create_authorization_header(service_id=sample_admin_service_id,
path=url_for('service.revoke_token', path=url_for('service.revoke_api_key',
service_id=sample_token.service_id), service_id=sample_api_key.service_id),
method='POST') method='POST')
response = client.post(url_for('service.revoke_token', service_id=sample_token.service_id), response = client.post(url_for('service.revoke_api_key', service_id=sample_api_key.service_id),
headers=[auth_header]) headers=[auth_header])
assert response.status_code == 202 assert response.status_code == 202
tokens_for_service = Token.query.filter_by(service_id=sample_token.service_id).first() api_keys_for_service = ApiKey.query.filter_by(service_id=sample_api_key.service_id).first()
assert tokens_for_service.expiry_date is not None assert api_keys_for_service.expiry_date is not None
def test_create_service_should_create_new_token_for_service(notify_api, notify_db, notify_db_session, sample_user, def test_create_service_should_create_new_service_for_user(notify_api, notify_db, notify_db_session, sample_user,
sample_admin_service_id): sample_admin_service_id):
with notify_api.test_request_context(): with notify_api.test_request_context():
with notify_api.test_client() as client: with notify_api.test_client() as client:
data = { data = {
@@ -394,12 +401,10 @@ def test_create_service_should_create_new_token_for_service(notify_api, notify_d
method='POST', method='POST',
request_body=json.dumps(data)) request_body=json.dumps(data))
headers = [('Content-Type', 'application/json'), auth_header] headers = [('Content-Type', 'application/json'), auth_header]
assert Token.query.count() == 1
resp = client.post(url_for('service.create_service'), resp = client.post(url_for('service.create_service'),
data=json.dumps(data), data=json.dumps(data),
headers=headers) headers=headers)
assert resp.status_code == 201 assert resp.status_code == 201
assert Token.query.count() == 2
def test_create_template(notify_api, notify_db, notify_db_session, sample_service, sample_admin_service_id): def test_create_template(notify_api, notify_db, notify_db_session, sample_service, sample_admin_service_id):