"""Synthetic HTTPS gateway tests. No production gateway or real password is used.""" import asyncio from contextlib import ExitStack import getpass import hashlib import json from pathlib import Path import platform import ssl import subprocess import sys import tempfile import threading import time import unittest from unittest.mock import patch sys.path.insert(0,str(Path(__file__).resolve().parents[1])) import aiohttp from aiohttp import web import auth_gate as a import gauntlet_v2 as g class PolicyTests(unittest.TestCase): def test_gateway_url_restrictions(self): bad=['http://trusted.example','https://evil.example','https://trusted.example/x', 'https://user:pass@trusted.example','https://trusted.example?x=1', 'https://trusted.example#secret','https://trusted.example:444'] for url in bad: with self.subTest(url=url),self.assertRaises(a.AuthenticationFailed): a.AdministratorGate({'gateway_hostname':'trusted.example','gateway_port':443},{'LIVE_AUTH_GATEWAY_URL':url}) def test_configured_local_https_gate_needs_no_cloudflare_hostname(self): gate=a.AdministratorGate({}, {'LIVE_AUTH_GATEWAY_URL':'https://localhost:8443/api/v1/auth/gate'}) self.assertEqual(gate.base,'https://localhost:8443') self.assertEqual(gate.hostname,'localhost') def test_bare_hostname_is_normalized(self): gate=a.AdministratorGate({'gateway_hostname':'trusted.example'}, {'LIVE_AUTH_GATEWAY_URL':'trusted.example'}) self.assertEqual(gate.base,'https://trusted.example') def test_partial_cloudflare_configuration_rejected(self): with self.assertRaises(a.AuthenticationFailed): a.AdministratorGate({'gateway_hostname':'trusted.example'}, {'LIVE_AUTH_GATEWAY_URL':'trusted.example','CF_ACCESS_CLIENT_ID':'synthetic'}) def test_mac_hidden_prompt_policy(self): gate=a.AdministratorGate({'gateway_hostname':'trusted.example'},{'LIVE_AUTH_GATEWAY_URL':'trusted.example'}) with patch.object(a.platform,'system',return_value='Linux'),self.assertRaises(a.AuthenticationFailed): gate._prompt() with patch.object(a.platform,'system',return_value='Darwin'),patch.object(a.sys.stdin,'isatty',return_value=False),self.assertRaises(a.AuthenticationFailed): gate._prompt() with patch.object(a.platform,'system',return_value='Darwin'),patch.object(a.sys.stdin,'isatty',return_value=True),\ patch.object(a.getpass,'getpass',side_effect=['synthetic password','123456']) as hidden,\ patch.object(a.resource,'setrlimit') as core: self.assertEqual(gate._prompt(),('synthetic password','123456')) self.assertEqual([call.args[0] for call in hidden.call_args_list], ['Enter App-Level Secret Password: ','Enter 6-Digit Google Authenticator MFA Token: ']) core.assert_called_once() def test_background_thread_cannot_prompt(self): gate=a.AdministratorGate({'gateway_hostname':'trusted.example'},{'LIVE_AUTH_GATEWAY_URL':'trusted.example'}) errors=[] def attempt(): try: gate._prompt() except a.AuthenticationFailed: errors.append(True) with patch.object(a.platform,'system',return_value='Darwin'),patch.object(a.sys.stdin,'isatty',return_value=True): t=threading.Thread(target=attempt);t.start();t.join() self.assertEqual(errors,[True]) class GatewayTests(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self): self.temp=tempfile.TemporaryDirectory() root=Path(self.temp.name) self.cert,self.key=root/'cert.pem',root/'key.pem' subprocess.run(['openssl','req','-x509','-newkey','rsa:2048','-nodes','-days','1', '-keyout',str(self.key),'-out',str(self.cert),'-subj','/CN=localhost', '-addext','subjectAltName=DNS:localhost'],check=True,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL) server_ssl=ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER);server_ssl.load_cert_chain(self.cert,self.key) self.client_ssl=ssl.create_default_context(cafile=str(self.cert)) self.requests=[];self.failures=[];self.status='AUTHENTICATED';self.expiry=3600 self.secure=True;self.http_only=True;self.cookie=True;self.redirect=False;self.bind=True app=web.Application();app.router.add_post('/api/v1/auth/gate',self.authenticate) app.router.add_post('/v1/broker-session',self.bootstrap) self.runner=web.AppRunner(app,access_log=None);await self.runner.setup() self.site=web.TCPSite(self.runner,'127.0.0.1',0,ssl_context=server_ssl);await self.site.start() self.port=self.site._server.sockets[0].getsockname()[1] self.seed='GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ' self.vault=g.LocalVault(root/'auth.vault.json') self.vault.create('SYNTHETIC-PASSWORD',self.seed) self.gate=a.AdministratorGate({'gateway_hostname':'localhost','gateway_port':self.port,'vault_path':str(self.vault.path)}, {'LIVE_AUTH_GATEWAY_URL':f'https://localhost:{self.port}'},self.failures.append) self.original_session=aiohttp.ClientSession self.patches=ExitStack() def session(**kwargs): kwargs.setdefault('connector',aiohttp.TCPConnector(ssl=self.client_ssl)) return self.original_session(**kwargs) self.patches.enter_context(patch.object(a.aiohttp,'ClientSession',side_effect=session)) self.prompt_counter=0 def next_factors(): counter=int(time.time())//30+self.prompt_counter self.prompt_counter+=1 return ('SYNTHETIC-PASSWORD',g.LocalVault.token(self.seed,counter)) self.prompt=self.patches.enter_context(patch.object(a.AdministratorGate,'_prompt',side_effect=next_factors)) async def asyncTearDown(self): await self.gate.close();self.patches.close();await self.runner.cleanup();self.temp.cleanup() async def authenticate(self,request): body=await request.json();self.requests.append(('gate',body)) if self.redirect:return web.Response(status=302,headers={'Location':'https://untrusted.invalid/'}) response=web.json_response({'status':self.status,'session_token_expires_sec':self.expiry}) if self.cookie:response.set_cookie('alpha_session','SYNTHETIC-SESSION',secure=self.secure,httponly=self.http_only,path='/') return response async def bootstrap(self,request): self.requests.append(('bootstrap',await request.json())) if request.cookies.get('alpha_session')!='SYNTHETIC-SESSION':return web.Response(status=403) body=self.requests[-1][1] return web.json_response({'bound':self.bind,'account_id':body['account_id']}) async def test_exact_challenge_and_cookie_bound_bootstrap(self): await self.gate.open() expected=hashlib.sha256(b'SYNTHETIC-PASSWORD').hexdigest() self.assertEqual(self.requests,[('gate',{'gate_challenge':expected})]) self.gate.require() self.assertGreater(self.gate.expires_in,3590) await self.gate.bootstrap('alpaca','SYNTHETIC-ACCOUNT','SYNTHETIC-KEY','SYNTHETIC-SECRET') self.assertEqual(self.gate.bound_account,'SYNTHETIC-ACCOUNT') self.assertEqual(self.requests[-1][1],{'broker':'alpaca','account_id':'SYNTHETIC-ACCOUNT', 'api_key':'SYNTHETIC-KEY','secret_key':'SYNTHETIC-SECRET'}) async def test_local_mfa_failure_never_contacts_gateway(self): self.prompt.side_effect=lambda: ('SYNTHETIC-PASSWORD','not-a-token') with self.assertRaises(a.AuthenticationFailed):await self.gate.open() self.assertEqual(self.requests,[]) async def test_authenticated_config_mismatch_opens_portal_instead_of_crashing(self): import io import os import socket from contextlib import redirect_stdout project=Path(self.temp.name)/'project' g.initialize(project,'alpaca','SPY');cfg=g.configuration(project) journal=g.Journal(project,cfg,'old-synthetic-key-fingerprint') journal.observe('PREFLIGHT',{'cash':g.D('500'),'equity':g.D('500'),'gross_asset_value':g.D('0'),'quantity':g.D('0')}) old_binding=journal.get('binding');journal.close() (project/'security.json').write_text(json.dumps({'gateway_hostname':'localhost','gateway_port':self.port, 'vault_path':str(self.vault.path),'account_id':'SYNTHETIC'})) with socket.socket() as sock:sock.bind(('127.0.0.1',0));port=sock.getsockname()[1] factors=('SYNTHETIC-PASSWORD',g.LocalVault.token(self.seed,int(time.time())//30)) capture=io.StringIO() with patch.object(g.AdministratorGate,'prompt_factors',return_value=factors), patch.object(g,'trading') as trading, patch.dict(os.environ,{'LIVE_AUTH_GATEWAY_URL':self.gate.base, 'LIVE_BROKER_API_KEY':'SYNTHETIC-NEW-KEY','LIVE_BROKER_SECRET_KEY':'SYNTHETIC-SECRET'},clear=True), redirect_stdout(capture): result=await asyncio.to_thread(g.main,['--live','--serve','--project',str(project),'--port',str(port),'--duration','1']) self.assertEqual(result,2) self.assertIn('LOCAL_BINDING_REVIEW_REQUIRED',capture.getvalue()) self.assertNotIn('STARTUP_FAILED',capture.getvalue());trading.assert_not_called() journal=g.Journal(project,cfg,'old-synthetic-key-fingerprint') try:self.assertEqual(journal.get('binding'),old_binding) finally:journal.close() async def test_wrong_status_hard_fails_once(self): self.status='approved' with self.assertRaises(a.AuthenticationFailed):await self.gate.open() self.assertEqual(len(self.requests),1);self.assertEqual(self.failures,['AUTHENTICATION_FAILED']) async def test_wrong_expiry_hard_fails(self): self.expiry=180 with self.assertRaises(a.AuthenticationFailed):await self.gate.open() self.assertEqual(len(self.requests),1) async def test_boolean_expiry_is_not_integer(self): self.expiry=True with self.assertRaises(a.AuthenticationFailed):await self.gate.open() async def test_missing_session_cookie_rejected(self): self.cookie=False with self.assertRaises(a.AuthenticationFailed):await self.gate.open() async def test_nonsecure_cookie_rejected(self): self.secure=False with self.assertRaises(a.AuthenticationFailed):await self.gate.open() async def test_non_http_only_cookie_rejected(self): self.http_only=False with self.assertRaises(a.AuthenticationFailed):await self.gate.open() async def test_redirect_is_not_followed(self): self.redirect=True with self.assertRaises(a.AuthenticationFailed):await self.gate.open() self.assertEqual(len(self.requests),1) async def test_background_revocation_and_fresh_prompt(self): await self.gate.open() self.gate.deadline=time.monotonic()-1 await asyncio.sleep(.15) self.assertTrue(self.gate.revoked.is_set()) with self.assertRaises(a.AuthenticationFailed):self.gate.require() await self.gate.renew_if_due() self.assertEqual(self.prompt.call_count,2) self.assertEqual(self.gate.generation,2) self.gate.require() async def test_failed_renewal_revokes_and_halts(self): await self.gate.open();self.gate.deadline=time.monotonic()-1;self.status='DENIED' with self.assertRaises(a.AuthenticationFailed):await self.gate.renew_if_due() self.assertTrue(self.gate.revoked.is_set());self.assertEqual(self.failures,['AUTHENTICATION_FAILED']) async def test_bad_account_bootstrap_revokes(self): await self.gate.open();self.bind=False with self.assertRaises(a.AuthenticationFailed):await self.gate.bootstrap('kraken','SYNTHETIC','SYNTHETIC','SYNTHETIC') with self.assertRaises(a.AuthenticationFailed):self.gate.require() async def test_live_broker_cannot_bypass_missing_gate(self): with tempfile.TemporaryDirectory() as folder: root=Path(folder);g.initialize(root,'alpaca','SPY');cfg=g.configuration(root) journal=g.Journal(root,cfg,'synthetic') try: with self.assertRaises(a.AuthenticationFailed): async with g.Alpaca(cfg,'SYNTHETIC','SYNTHETIC',journal):pass finally:journal.close()