AdamWen95
83 lines
3.6 KiB
Python
83 lines
3.6 KiB
Python
import importlib.util
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
import contextlib
|
|
import io
|
|
import json
|
|
import sys
|
|
from types import SimpleNamespace
|
|
|
|
MODULE = Path(__file__).resolve().parents[1] / "deploy/start.py"
|
|
spec = importlib.util.spec_from_file_location("start_console", MODULE)
|
|
start = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(start)
|
|
|
|
|
|
class DeploymentPathsTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.base = Path(self.temp.name)
|
|
self.project = self.base / "project"
|
|
self.project.mkdir()
|
|
self.outside = self.base / "other-owner"
|
|
self.outside.mkdir()
|
|
|
|
def validator(self):
|
|
self.assertTrue(hasattr(start, "validate_runtime"), "runtime boundary validation is missing")
|
|
return start.validate_runtime
|
|
|
|
def link(self, target, destination):
|
|
try:
|
|
destination.symlink_to(target, target_is_directory=True)
|
|
except OSError:
|
|
self.skipTest("OS account cannot create a directory symlink")
|
|
|
|
def test_runtime_parent_link_cannot_escape_project(self):
|
|
validate = self.validator()
|
|
self.link(self.outside, self.project / "runtime")
|
|
before = self.outside.stat().st_mode
|
|
with self.assertRaises(ValueError):
|
|
validate(self.project)
|
|
self.assertEqual(self.outside.stat().st_mode, before)
|
|
self.assertEqual(list(self.outside.iterdir()), [])
|
|
|
|
def test_runtime_child_link_cannot_escape_project(self):
|
|
validate = self.validator()
|
|
(self.project / "runtime").mkdir()
|
|
self.link(self.outside, self.project / "runtime/data")
|
|
with self.assertRaises(ValueError):
|
|
validate(self.project)
|
|
self.assertEqual(list(self.outside.iterdir()), [])
|
|
|
|
def test_normal_runtime_is_resolved_without_writing(self):
|
|
validate = self.validator()
|
|
self.assertEqual(validate(self.project), self.project.resolve() / "runtime")
|
|
self.assertEqual(list(self.project.iterdir()), [])
|
|
|
|
def test_backend_is_only_on_proxy_loopback_namespace(self):
|
|
# Regress the reviewed direct-container-IP role-header bypass at the Docker boundary.
|
|
deploy = self.project / 'deploy'
|
|
deploy.mkdir()
|
|
(deploy/'lan-settings.json').write_text(json.dumps({
|
|
'address':'10.88.118.188','allowed_cidrs':['10.88.0.0/17','10.20.0.0/21'],'port':14440}))
|
|
calls = []
|
|
def docker(args, **kwargs):
|
|
calls.append(args)
|
|
absent = args[:3] in (['docker','container','inspect'], ['docker','network','inspect'])
|
|
return SimpleNamespace(returncode=1 if absent else 0, stdout='')
|
|
with mock.patch.object(sys, 'argv', ['start.py','--project-root',str(self.project)]), \
|
|
mock.patch.object(sys, 'path', [str(MODULE.parent)] + sys.path), \
|
|
mock.patch.object(start.subprocess, 'run', side_effect=docker), \
|
|
mock.patch.object(start.os, 'chown', create=True), \
|
|
mock.patch.object(start.socket, 'socket'), contextlib.redirect_stdout(io.StringIO()):
|
|
start.main()
|
|
backend = next(c for c in calls if '--name' in c and c[c.index('--name')+1]=='adamwen-test-console')
|
|
proxy = next(c for c in calls if '--name' in c and c[c.index('--name')+1]=='adamwen-test-console-proxy')
|
|
self.assertIn('RUNDECK_SERVER_ADDRESS=127.0.0.1', backend)
|
|
self.assertEqual(backend[backend.index('--network')+1], 'container:adamwen-test-console-proxy')
|
|
self.assertNotIn('--publish', backend)
|
|
self.assertLess(calls.index(proxy), calls.index(backend))
|