70 lines
4.5 KiB
JavaScript
70 lines
4.5 KiB
JavaScript
import test from 'node:test'
|
|
import assert from 'node:assert/strict'
|
|
import { spawnSync } from 'node:child_process'
|
|
import { fileURLToPath } from 'node:url'
|
|
|
|
test('offline proxy routing/authentication/TTL/transparency and redaction (no sockets or HTTP)', () => {
|
|
const script = String.raw`
|
|
import importlib.util,json,sys,unittest
|
|
spec=importlib.util.spec_from_file_location('proxy',sys.argv[1]);p=importlib.util.module_from_spec(spec);spec.loader.exec_module(p)
|
|
class ProxyTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.now=[10.0];self.calls=[];self.logs=[]
|
|
self.config={'customerId':'7','controlToken':'private-control','internalToken':'private-internal-token',
|
|
'orderIps':['172.20.0.10'],'requestIds':['fixture-request'],'upstreams':{'IAM':'http://iam.invalid:9201','FINANCE':'http://finance.invalid:9207'}}
|
|
def forward(*args): self.calls.append(args);return 201,{'X-Upstream':'kept'},b'original-response'
|
|
self.router=p.Router(self.config,forward=forward,clock=lambda:self.now[0],record=self.logs.append)
|
|
self.headers={'X-Internal-Service-Token':'private-internal-token','X-Request-Id':'fixture-request'}
|
|
self.path='/iam/internal/iam/tenants/7/order-eligibility'
|
|
def send(self,path=None,method='GET',headers=None,peer='172.20.0.10'):
|
|
return self.router.dispatch(method,path or self.path,headers or self.headers,b'original-body',peer)
|
|
def test_default_pass_and_transport_are_unchanged(self):
|
|
self.assertEqual(self.send(),(201,{'X-Upstream':'kept'},b'original-response'))
|
|
self.assertEqual(self.calls[0][2],'/internal/iam/tenants/7/order-eligibility')
|
|
self.assertEqual(self.calls[0][3]['X-Internal-Service-Token'],'private-internal-token')
|
|
self.assertEqual(self.calls[0][4],b'original-body')
|
|
def test_only_exact_authenticated_customer_get_from_order_faults(self):
|
|
self.router.set_mode('private-control','IAM',60)
|
|
self.assertEqual(self.send()[0],503);self.assertEqual(len(self.calls),0)
|
|
for path,method,headers,peer in [
|
|
(self.path.replace('/7/','/8/'),'GET',self.headers,'172.20.0.10'),
|
|
(self.path+'?other=1','GET',self.headers,'172.20.0.10'),
|
|
(self.path,'POST',self.headers,'172.20.0.10'),
|
|
(self.path,'GET',{'X-Internal-Service-Token':'wrong'},'172.20.0.10'),
|
|
(self.path,'GET',{**self.headers,'X-Request-Id':'other-order-request'},'172.20.0.10'),
|
|
(self.path,'GET',self.headers,'172.20.0.11'),
|
|
('/finance/internal/finance/v1/wallet/accounts/7','GET',self.headers,'172.20.0.10'),
|
|
('/iam/internal/iam/finance-customers?page=1','GET',self.headers,'172.20.0.10')]:
|
|
self.assertEqual(self.send(path,method,headers,peer)[0],201)
|
|
def test_expiry_needs_no_control_request_and_returns_to_pass(self):
|
|
self.router.set_mode('private-control','FINANCE',5)
|
|
path='/finance/internal/finance/v1/wallet/accounts/7'
|
|
self.assertEqual(self.send(path)[0],503)
|
|
self.now[0]=15.0
|
|
self.assertEqual(self.send(path)[0],201);self.assertEqual(self.router.status()['mode'],'PASS')
|
|
def test_control_auth_and_malformed_authenticated_command(self):
|
|
self.router.set_mode('private-control','IAM',10)
|
|
with self.assertRaises(PermissionError):self.router.set_mode('wrong','PASS')
|
|
for mode,ttl in [('UNKNOWN',10),('IAM',0),('IAM',61),('IAM',True)]:
|
|
with self.assertRaises(ValueError):self.router.set_mode('private-control',mode,ttl)
|
|
self.assertEqual(self.router.status()['mode'],'PASS')
|
|
def test_evidence_contains_no_secret_headers_or_payload(self):
|
|
self.router.set_mode('private-control','IAM',10);self.send()
|
|
serialized=json.dumps(self.logs)
|
|
for secret in ['private-control','private-internal-token','original-body']:self.assertNotIn(secret,serialized)
|
|
self.assertTrue(self.logs[0]['internalTokenValidated']);self.assertFalse(self.logs[0]['forwarded'])
|
|
self.assertEqual(self.logs[0]['requestId'],'fixture-request')
|
|
def test_hop_headers_removed_but_identity_preserved(self):
|
|
headers=p.forward_headers({'connection':'x-hop','x-hop':'drop','Host':'old','X-Internal-Service-Token':'keep','Content-Length':'99'},3)
|
|
self.assertNotIn('x-hop',headers);self.assertNotIn('Host',headers)
|
|
self.assertEqual(headers['X-Internal-Service-Token'],'keep');self.assertEqual(headers['Content-Length'],'3')
|
|
result=unittest.TextTestRunner().run(unittest.defaultTestLoader.loadTestsFromTestCase(ProxyTests))
|
|
sys.exit(0 if result.wasSuccessful() else 1)
|
|
`
|
|
const result = spawnSync(process.env.PYTHON || 'python', ['-c', script,
|
|
fileURLToPath(new URL('./order-submit-dependency-fault-proxy.py', import.meta.url))],
|
|
{ encoding:'utf8', env:{...process.env,PYTHONDONTWRITEBYTECODE:'1'} })
|
|
assert.equal(result.status,0,result.stderr || result.error?.message)
|
|
assert.match(result.stderr,/Ran 6 tests/)
|
|
})
|