lfg2025/archy
1
0
Fork 0
Code Issues 11 Pull Requests Actions Packages Projects Releases 12 Wiki Activity

Complete private gateway and local website publishing UAT

Browse Source
This commit is contained in:
archipelago
2026-10-08 18:10:41 -04:00
parent 768e828246
commit 3ab4162a8b
38 changed files with 2232 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/public-web-gateway/test_enroll.py
+32
View File
@@ -0,0 +1,32 @@
import json
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
SCRIPT = Path(__file__).resolve().parents[2] / 'scripts/public-web-gateway/enroll.py'
class EnrollmentTests(unittest.TestCase):
def test_private_export_duplicate_domain_and_explicit_rotation(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
subprocess.run(['openssl', 'req', '-x509', '-newkey', 'rsa:2048', '-nodes', '-keyout', str(root/'key'), '-out', str(root/'ca'), '-days', '1', '-subj', '/CN=gateway.example'], check=True, capture_output=True)
config = {'bindPort': 7400, 'auth': {'method': 'token', 'token': 't'*64}, 'transport': {'tls': {'force': True}}, 'httpPlugins': [{'ops': ['Login', 'NewProxy', 'Ping', 'NewWorkConn', 'NewUserConn']}]}
path = root/'frps.json'; path.write_text(json.dumps(config)); path.chmod(0o600)
base = [sys.executable, str(SCRIPT), '--frps-config', str(path), '--policy', str(root/'policy.json'), '--ca', str(root/'ca'), '--host', 'gateway.example', '--tls-server-name', 'gateway.example', '--domain', 'site.example']
result = subprocess.run(base + ['--name', 'node-a', '--output', str(root/'node-a.json')], capture_output=True)
self.assertEqual(result.returncode, 0, result.stderr.decode())
private = json.loads((root/'node-a.json').read_text())
self.assertNotIn(private['enrollment_token'].encode(), result.stdout + result.stderr)
self.assertEqual((root/'node-a.json').stat().st_mode & 0o777, 0o600)
first = (root/'policy.json').read_bytes()
result = subprocess.run(base + ['--name', 'node-b', '--output', str(root/'node-b.json')], capture_output=True)
self.assertNotEqual(result.returncode, 0)
self.assertEqual(first, (root/'policy.json').read_bytes())
self.assertFalse((root/'node-b.json').exists())
result = subprocess.run(base + ['--name', 'node-a', '--rotate', '--output', str(root/'rotated.json')], capture_output=True)
self.assertEqual(result.returncode, 0, result.stderr.decode())
self.assertNotEqual(private['enrollment_token'], json.loads((root/'rotated.json').read_text())['enrollment_token'])
self.assertNotEqual(first, (root/'policy.json').read_bytes())
if __name__ == '__main__': unittest.main()
tests/public-web-gateway/test_policy.py
+45
View File
@@ -0,0 +1,45 @@
import hashlib
import importlib.util
from pathlib import Path
import unittest
spec = importlib.util.spec_from_file_location('gateway_policy', Path(__file__).resolve().parents[2] / 'scripts/public-web-gateway/policy.py')
policy = importlib.util.module_from_spec(spec)
spec.loader.exec_module(policy)
class PolicyTests(unittest.TestCase):
def setUp(self):
self.token = 'synthetic-test-token-' * 3
self.user = {'user': 'framework', 'metas': {'enrollment_token': self.token}}
self.entries = {'framework': {'enabled': True, 'token_sha256': hashlib.sha256(self.token.encode()).hexdigest(), 'domains': ['free.archipelago.builders']}}
self.proxy = {'user': self.user, 'proxy_name': 'framework.website', 'proxy_type': 'https', 'custom_domains': ['free.archipelago.builders']}
def test_allow_assigned_https_only(self):
self.assertTrue(policy.authorize('Login', self.user, self.entries))
for op in ('NewProxy', 'NewUserConn', 'Ping', 'NewWorkConn'):
self.assertTrue(policy.authorize(op, self.proxy, self.entries))
def test_revoke_and_rotate_apply_to_all_operations(self):
for op in policy.OPS:
content = self.user if op == 'Login' else self.proxy
self.entries['framework']['enabled'] = False
self.assertFalse(policy.authorize(op, content, self.entries))
self.entries['framework']['enabled'] = True
self.entries['framework']['token_sha256'] = '0' * 64
self.assertFalse(policy.authorize(op, content, self.entries))
def test_no_other_domains_protocols_or_shared_groups(self):
for key, value in [('custom_domains', ['other.example']), ('custom_domains', ['free.archipelago.builders', 'other.example']), ('proxy_type', 'tcp'), ('proxy_type', 'http'), ('remote_port', 22), ('subdomain', 'admin'), ('group', 'shared'), ('locations', ['/']), ('proxy_name', 'another.website')]:
with self.subTest(key=key, value=value):
self.assertFalse(policy.authorize('NewProxy', {**self.proxy, key: value}, self.entries))
def test_missing_or_malformed_auth_is_denied(self):
for user in ({}, {'user': 'framework'}, {'user': 'framework', 'metas': []}, {'user': 'framework', 'metas': {'enrollment_token': 'wrong'}}):
self.assertFalse(policy.authorize('Login', user, self.entries))
self.assertFalse(policy.authorize('Unknown', self.proxy, self.entries))
self.assertFalse(policy.authorize('Login', self.user, {}))
if __name__ == '__main__':
unittest.main()
tests/public-web-gateway/test_router.py
+53
View File
@@ -0,0 +1,53 @@
import importlib.util
from pathlib import Path
import unittest
spec = importlib.util.spec_from_file_location('node_router', Path(__file__).resolve().parents[2] / 'docker/public-web-router/router.py')
router = importlib.util.module_from_spec(spec)
spec.loader.exec_module(router)
class RouterTests(unittest.TestCase):
def setUp(self):
self.config = {'schema': 1, 'gateway': {'host': '192.0.2.1', 'port': 7400, 'node_id': 'node-a', 'transport_token': 'a'*64, 'enrollment_token': 'b'*64, 'ca_pem': '-----BEGIN CERTIFICATE-----\nexample\n-----END CERTIFICATE-----', 'tls_server_name': 'gateway.example', 'domains': ['site.example']}, 'routes': [{'id': 'site-a', 'domain': 'site.example', 'fips_address': 'fd00::1', 'port': 32000}]}
def test_tls_ends_on_node_with_pinned_control_channel(self):
caddy, frpc, pem = router.render(self.config)
self.assertIn('disable_http_challenge', caddy)
self.assertNotIn('tls internal', caddy)
self.assertIn('bind 127.0.0.1', caddy)
self.assertEqual(frpc['proxies'][0]['type'], 'https')
self.assertEqual(frpc['transport']['tls']['serverName'], 'gateway.example')
self.assertIn('trustedCaFile', frpc['transport']['tls'])
def test_refuses_management_and_arbitrary_upstreams(self):
for port in (22, 443, 7474, 8191, 31999, 32032, True):
self.config['routes'][0]['port'] = port
with self.assertRaises(ValueError): router.render(self.config)
self.config['routes'][0]['port'] = 32000
for address in ('127.0.0.1', '::1', '2001:db8::1'):
self.config['routes'][0]['fips_address'] = address
with self.assertRaises(ValueError): router.render(self.config)
def test_refuses_config_injection_and_duplicate_domains(self):
for key in ('domain', 'id'):
old = self.config['routes'][0][key]
self.config['routes'][0][key] = 'site.example\n import /secret'
with self.assertRaises(ValueError): router.render(self.config)
self.config['routes'][0][key] = old
self.config['routes'].append(dict(self.config['routes'][0]))
with self.assertRaises(ValueError): router.render(self.config)
def test_app_routes_keep_the_expected_app_gate_identity(self):
self.config['routes'][0].update(id='app-photoprism', app_id='photoprism', port=2342)
caddy, _, _ = router.render(self.config)
self.assertIn('header_up X-Archipelago-App photoprism', caddy)
self.config['routes'][0]['id'] = 'app-another'
with self.assertRaises(ValueError): router.render(self.config)
def test_test_certificates_require_explicit_mode(self):
self.config['certificate_mode'] = 'test'
self.assertIn('tls internal', router.render(self.config)[0])
self.config['certificate_mode'] = 'insecure'
with self.assertRaises(ValueError): router.render(self.config)
if __name__ == '__main__': unittest.main()