vbatts/quay
Archived
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Add delete logging and tests for logging

Browse source
This commit is contained in:
Joseph Schorr 2016-04-14 15:04:32 -04:00 • committed by Jimmy Zelinskie
parent 6aa7040f39
commit 5d6e5a42e8
2 changed files with 39 additions and 20 deletions
Download patch file Download diff file Expand all files Collapse all files

10
endpoints/key_server.py
View file

@ -199,6 +199,16 @@ def delete_service_key(service, kid):
except data.model.ServiceKeyDoesNotExist: except data.model.ServiceKeyDoesNotExist:
abort(404) abort(404)
key_log_metadata = {
'kid': kid,
'signer_kid': signer_key.kid,
'service': service,
'name': signer_key.name,
'user_agent': request.headers.get('User-Agent'),
'ip': request.remote_addr,
}
log_action('service_key_delete', None, metadata=key_log_metadata, ip=request.remote_addr)
return make_response('', 204) return make_response('', 204)
abort(403) abort(403)

49
test/test_endpoints.py
View file

@ -21,6 +21,7 @@ from endpoints.api import api, api_bp
from endpoints.api.user import Signin from endpoints.api.user import Signin
from endpoints.web import web as web_bp from endpoints.web import web as web_bp
from initdb import setup_database_for_testing, finished_database_for_testing from initdb import setup_database_for_testing, finished_database_for_testing
from test.helpers import assert_action_logged
try: try:
@ -205,8 +206,9 @@ class KeyServerTestCase(EndpointTestCase):
jwkset = py_json.loads(rv) jwkset = py_json.loads(rv)
# Make sure the unapproved key isn't returned in our results # Make sure the unapproved key isn't returned in our results
self.assertTrue(len(jwkset['keys']) > 0)
for jwk in jwkset['keys']: for jwk in jwkset['keys']:
self.assertTrue(jwk != unapproved_key.jwk) self.assertNotEquals(jwk, unapproved_key.jwk)
def test_get_service_key(self): def test_get_service_key(self):
# 200 for an approved key # 200 for an approved key
@ -232,19 +234,25 @@ class KeyServerTestCase(EndpointTestCase):
token = jwt.encode(payload, private_key.exportKey('PEM'), 'RS256') token = jwt.encode(payload, private_key.exportKey('PEM'), 'RS256')
# Publish a new key # Publish a new key
self.putResponse('key_server.put_service_key', service='sample_service', kid='kid420', with assert_action_logged('service_key_create'):
headers={ self.putResponse('key_server.put_service_key', service='sample_service', kid='kid420',
'Authorization': 'Bearer %s' % token, headers={
'Content-Type': 'application/json', 'Authorization': 'Bearer %s' % token,
}, data=jwk, expected_code=202) 'Content-Type': 'application/json',
}, data=jwk, expected_code=202)
# Ensure that the key exists but is unapproved.
self.getResponse('key_server.get_service_key', service='sample_service', kid='kid420',
expected_code=409)
# Rotate that new key # Rotate that new key
token = jwt.encode(payload, private_key.exportKey('PEM'), 'RS256', headers={'kid': 'kid420'}) with assert_action_logged('service_key_rotate'):
self.putResponse('key_server.put_service_key', service='sample_service', kid='kid6969', token = jwt.encode(payload, private_key.exportKey('PEM'), 'RS256', headers={'kid': 'kid420'})
headers={ self.putResponse('key_server.put_service_key', service='sample_service', kid='kid6969',
'Authorization': 'Bearer %s' % token, headers={
'Content-Type': 'application/json', 'Authorization': 'Bearer %s' % token,
}, data=jwk, expected_code=200) 'Content-Type': 'application/json',
}, data=jwk, expected_code=200)
# Rotation should only work when signed by the previous key # Rotation should only work when signed by the previous key
private_key = RSA.generate(2048) private_key = RSA.generate(2048)
@ -256,6 +264,7 @@ class KeyServerTestCase(EndpointTestCase):
'Content-Type': 'application/json', 'Content-Type': 'application/json',
}, data=jwk, expected_code=403) }, data=jwk, expected_code=403)
def test_delete_service_key(self): def test_delete_service_key(self):
# No Authorization header should yield a 400 # No Authorization header should yield a 400
self.deleteResponse('key_server.delete_service_key', expected_code=400, self.deleteResponse('key_server.delete_service_key', expected_code=400,
@ -271,9 +280,10 @@ class KeyServerTestCase(EndpointTestCase):
headers={'kid': 'kid123'}) headers={'kid': 'kid123'})
# Using the credentials of our approved key, delete our unapproved key # Using the credentials of our approved key, delete our unapproved key
self.deleteResponse('key_server.delete_service_key', with assert_action_logged('service_key_delete'):
headers={'Authorization': 'Bearer %s' % token}, self.deleteResponse('key_server.delete_service_key',
expected_code=204, service='sample_service', kid='kid321') headers={'Authorization': 'Bearer %s' % token},
expected_code=204, service='sample_service', kid='kid321')
# Attempt to delete a key signed by a key from a different service # Attempt to delete a key signed by a key from a different service
bad_token = jwt.encode(self._get_test_jwt_payload(), private_key.exportKey('PEM'), 'RS256', bad_token = jwt.encode(self._get_test_jwt_payload(), private_key.exportKey('PEM'), 'RS256',
@ -283,11 +293,10 @@ class KeyServerTestCase(EndpointTestCase):
expected_code=403, service='sample_service', kid='kid123') expected_code=403, service='sample_service', kid='kid123')
# Delete a self-signed, approved key # Delete a self-signed, approved key
self.deleteResponse('key_server.delete_service_key', with assert_action_logged('service_key_delete'):
headers={'Authorization': 'Bearer %s' % token}, self.deleteResponse('key_server.delete_service_key',
expected_code=204, service='sample_service', kid='kid123') headers={'Authorization': 'Bearer %s' % token},
expected_code=204, service='sample_service', kid='kid123')
if __name__ == '__main__': if __name__ == '__main__':