This repository has been archived on 2020-03-24. You can view files and clone it, but cannot push or open issues or pull requests.
quay/data/model/service_keys.py
2016-04-29 13:38:25 -04:00

144 lines
4.7 KiB
Python

from calendar import timegm
from datetime import datetime, timedelta
from peewee import JOIN_LEFT_OUTER
from app import app
from data.database import db_for_update, User, ServiceKey, ServiceKeyApproval
from data.model import ServiceKeyDoesNotExist, ServiceKeyAlreadyApproved, db_transaction
from data.model.notification import create_notification, delete_all_notifications_by_path_prefix
# TODO ACTION_LOGS for keys
UNAPPROVED_TTL = timedelta(seconds=app.config['UNAPPROVED_SERVICE_KEY_TTL_SEC'])
def _expired_keys_clause(service):
return ((ServiceKey.service == service) &
(ServiceKey.expiration_date <= datetime.utcnow()))
def _stale_unapproved_keys_clause(service):
return ((ServiceKey.service == service) &
(ServiceKey.approval >> None) &
(ServiceKey.created_date <= (datetime.utcnow() - UNAPPROVED_TTL)))
def _gc_expired(service):
ServiceKey.delete().where(_expired_keys_clause(service) |
_stale_unapproved_keys_clause(service)).execute()
def _notify_superusers(key):
notification_metadata = {
'name': key.name,
'kid': key.kid,
'service': key.service,
'jwk': key.jwk,
'metadata': key.metadata,
'created_date': timegm(key.created_date.utctimetuple()),
'expiration_date': timegm(key.created_date.utctimetuple()),
}
superusers = User.select().where(User.username << app.config['SUPER_USERS'])
for superuser in superusers:
create_notification('service_key_submitted', superuser, metadata=notification_metadata,
lookup_path='/service_key_approval/{0}'.format(key.kid))
def create_service_key(name, kid, service, jwk, metadata, expiration_date):
key = ServiceKey.create(name=name, kid=kid, service=service, jwk=jwk, metadata=metadata,
expiration_date=expiration_date)
_notify_superusers(key)
_gc_expired(service)
def replace_service_key(old_kid, kid, jwk, metadata, expiration_date):
try:
with db_transaction():
key = db_for_update(ServiceKey.select().where(ServiceKey.kid == old_kid)).get()
metadata = key.metadata.update(metadata)
ServiceKey.create(name=key.name, kid=kid, service=key.service, jwk=jwk,
metadata=metadata, expiration_date=expiration_date, approval=key.approval)
key.delete_instance()
except ServiceKey.DoesNotExist:
raise ServiceKeyDoesNotExist
_notify_superusers(key)
delete_all_notifications_by_path_prefix('/service_key_approval/{0}'.format(old_kid))
_gc_expired(key.service)
def update_service_key(name, kid, metadata, expiration_date):
try:
with db_transaction():
key = db_for_update(ServiceKey.select().where(ServiceKey.kid == kid)).get()
key.name = name
key.metadata.update(metadata)
key.expiration_date = expiration_date
key.save()
except ServiceKey.DoesNotExist:
raise ServiceKeyDoesNotExist
_gc_expired(key.service)
def delete_service_key(service, kid):
try:
ServiceKey.delete().where(ServiceKey.service == service,
ServiceKey.kid == kid).execute()
except ServiceKey.DoesNotExist:
raise ServiceKeyDoesNotExist()
delete_all_notifications_by_path_prefix('/service_key_approval/{0}'.format(kid))
_gc_expired(service)
def approve_service_key(kid, approver, approval_type, notes=''):
try:
with db_transaction():
key = db_for_update(ServiceKey.select().where(ServiceKey.kid == kid)).get()
if key.approval is not None:
raise ServiceKeyAlreadyApproved
approval = ServiceKeyApproval.create(approver=approver, approval_type=approval_type,
notes=notes)
key.approval = approval
key.save()
except ServiceKey.DoesNotExist:
raise ServiceKeyDoesNotExist
delete_all_notifications_by_path_prefix('/service_key_approval/{0}'.format(kid))
_gc_expired(key.service)
def _list_service_keys_query(kid=None, service=None, approved_only=False):
query = ServiceKey.select().join(ServiceKeyApproval, JOIN_LEFT_OUTER)
if approved_only:
query = query.where(~(ServiceKey.approval >> None))
if service is not None:
query = query.where(ServiceKey.service == service)
query = query.where(~(_expired_keys_clause(service)) |
~(_stale_unapproved_keys_clause(service)))
if kid is not None:
query = query.where(ServiceKey.kid == kid)
return query
def list_keys():
return list(_list_service_keys_query())
def list_service_keys(service):
return list(_list_service_keys_query(service=service, approved_only=True))
def get_service_key(kid, service=None):
try:
return _list_service_keys_query(kid=kid, service=service).get()
except ServiceKey.DoesNotExist:
raise ServiceKeyDoesNotExist