111 lines
5.5 KiB
Python
111 lines
5.5 KiB
Python
"""Boundary regressions for receipt-scoped cleanup and credential forwarding."""
|
|
import importlib.util
|
|
import contextlib
|
|
import hashlib
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from unittest import mock
|
|
import urllib.request
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
spec = importlib.util.spec_from_file_location('routes', Path(__file__).with_name('accept_forge_routes.py'))
|
|
m = importlib.util.module_from_spec(spec); spec.loader.exec_module(m)
|
|
|
|
|
|
class CleanupTests(unittest.TestCase):
|
|
def test_replaced_or_changed_repository_is_not_deleted(self):
|
|
for change in ({'id': 18}, {'private': False}, {'description': 'changed'},
|
|
{'owner': {'login': 'someone-else'}}):
|
|
with self.subTest(change=change), tempfile.TemporaryDirectory() as directory:
|
|
name = 'custodian-routes-' + 'a' * 24
|
|
record = dict(id=17, name=name, private=True, description=name, owner={'login': 'tegwick'})
|
|
record.update(change)
|
|
class Backend:
|
|
user = 'tegwick'
|
|
scope = 'test'
|
|
def api(self, method, *args):
|
|
self_outer.assertEqual(method, 'GET', 'must not delete changed ownership')
|
|
return 200, json.dumps(record).encode()
|
|
self_outer = self
|
|
suite = m.Suite(Backend(), Path(directory) / 'receipt.json', name)
|
|
suite.created_id = 17
|
|
with self.assertRaises(ValueError):
|
|
suite.close()
|
|
|
|
def test_ambiguous_creation_never_deletes_by_name(self):
|
|
class Backend:
|
|
user = 'tegwick'
|
|
scope = 'test'
|
|
def api(self, *args):
|
|
raise AssertionError('unowned creation must not issue API calls')
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
suite = m.Suite(Backend(), Path(directory) / 'receipt.json', 'custodian-routes-' + 'a' * 24)
|
|
suite.evidence['creation_attempted'] = True
|
|
suite.close()
|
|
self.assertTrue(suite.evidence['creation_outcome_requires_reconciliation'])
|
|
self.assertEqual(suite.evidence['result'], 'unverified')
|
|
|
|
def test_generic_error_is_not_detection_evidence(self):
|
|
self.assertFalse(m.detected(b'{"message":"internal error"}'))
|
|
self.assertFalse(m.detected(b'{"status":"unverified"}'))
|
|
|
|
|
|
class TransportTests(unittest.TestCase):
|
|
def test_http_redirect_does_not_forward_injected_credential(self):
|
|
hits = []
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
hits.append(self.path)
|
|
self.send_response(302)
|
|
self.send_header('Location', '/receiver')
|
|
self.end_headers()
|
|
def log_message(self, *args):
|
|
pass
|
|
server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
|
|
worker = threading.Thread(target=server.serve_forever, daemon=True); worker.start()
|
|
try:
|
|
backend = object.__new__(m.Production)
|
|
backend.origin = 'http://127.0.0.1:' + str(server.server_port)
|
|
backend.token = 'unbound-test-value'
|
|
backend.opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), m.NoRedirect())
|
|
code, _ = backend.api('GET', '/api/v1/user')
|
|
self.assertEqual(code, 302)
|
|
self.assertEqual(hits, ['/api/v1/user'])
|
|
finally:
|
|
server.shutdown(); server.server_close(); worker.join()
|
|
|
|
def test_handoff_is_silent_and_rejects_changed_source_before_fetch(self):
|
|
spec = importlib.util.spec_from_file_location('handoff', Path(__file__).with_name('run_forge_routes.py'))
|
|
wrapper = importlib.util.module_from_spec(spec); spec.loader.exec_module(wrapper)
|
|
for changed in (False, True):
|
|
with self.subTest(changed=changed), tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
proposal = {'repository_name': 'custodian-routes-' + 'a' * 24, 'source_sha256': {
|
|
n: hashlib.sha256((wrapper.HERE / n).read_bytes()).hexdigest()
|
|
for n in ('run_forge_routes.py', 'accept_forge_routes.py')}}
|
|
if changed:
|
|
proposal['source_sha256']['accept_forge_routes.py'] = '0' * 64
|
|
(root / 'proposal.json').write_text(json.dumps(proposal))
|
|
out, err = io.StringIO(), io.StringIO()
|
|
with mock.patch('sys.argv', ['handoff', '--proposal', str(root / 'proposal.json'),
|
|
'--receipt', str(root / 'result.json')]), mock.patch.object(
|
|
wrapper.subprocess, 'run', return_value=mock.Mock(returncode=5,
|
|
stdout=b'UNBOUND-CREDENTIAL-MARKER', stderr=b'UNBOUND-CREDENTIAL-MARKER')) as child:
|
|
with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
|
|
result = wrapper.main()
|
|
self.assertEqual(out.getvalue() + err.getvalue(), '')
|
|
if changed:
|
|
child.assert_not_called()
|
|
self.assertEqual(result, 2)
|
|
else:
|
|
self.assertEqual(result, 5)
|
|
self.assertTrue(child.call_args.kwargs['capture_output'])
|
|
self.assertNotIn('UNBOUND-CREDENTIAL-MARKER', (root / 'result.transport.json').read_text())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|