mirror of
https://github.com/GSA/notifications-api.git
synced 2026-01-02 14:10:46 -05:00
acee87fc63
The only update we should be doing to an api key is to expire/revoke the api key. Removed the update_dict from the the save method. Added an expire_api_key method that only updates the api key with an expiry date.
65 lines
1.8 KiB
Python
65 lines
1.8 KiB
Python
import uuid
|
|
from datetime import datetime
|
|
|
|
from flask import current_app
|
|
from itsdangerous import URLSafeSerializer
|
|
|
|
from app import db
|
|
from app.models import ApiKey
|
|
|
|
from app.dao.dao_utils import (
|
|
transactional,
|
|
version_class
|
|
)
|
|
|
|
|
|
@transactional
|
|
@version_class(ApiKey)
|
|
def save_model_api_key(api_key):
|
|
if not api_key.id:
|
|
api_key.id = uuid.uuid4() # must be set now so version history model can use same id
|
|
api_key.secret = _generate_secret()
|
|
db.session.add(api_key)
|
|
|
|
|
|
@transactional
|
|
@version_class(ApiKey)
|
|
def expire_api_key(service_id, api_key_id):
|
|
api_key = ApiKey.query.filter_by(id=api_key_id, service_id=service_id).one()
|
|
api_key.expiry_date = datetime.utcnow()
|
|
db.session.add(api_key)
|
|
|
|
|
|
def get_model_api_keys(service_id, id=None):
|
|
if id:
|
|
return ApiKey.query.filter_by(id=id, service_id=service_id, expiry_date=None).one()
|
|
return ApiKey.query.filter_by(service_id=service_id).all()
|
|
|
|
|
|
def get_unsigned_secrets(service_id):
|
|
"""
|
|
This method can only be exposed to the Authentication of the api calls.
|
|
"""
|
|
api_keys = ApiKey.query.filter_by(service_id=service_id, expiry_date=None).all()
|
|
keys = [_get_secret(x.secret) for x in api_keys]
|
|
return keys
|
|
|
|
|
|
def get_unsigned_secret(key_id):
|
|
"""
|
|
This method can only be exposed to the Authentication of the api calls.
|
|
"""
|
|
api_key = ApiKey.query.filter_by(id=key_id, expiry_date=None).one()
|
|
return _get_secret(api_key.secret)
|
|
|
|
|
|
def _generate_secret():
|
|
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'))
|