175 lines
7.9 KiB
Python
175 lines
7.9 KiB
Python
"""Exercise installer transformations and rollback against local fixture files."""
|
|
import contextlib
|
|
import importlib.util
|
|
import io
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
|
|
SOURCE = Path(__file__).resolve().parents[1]/'install-corp-proxy-security.py'
|
|
LOCATIONS = '''
|
|
location /ws/ {
|
|
proxy_pass http://eusrr_backend;
|
|
proxy_set_header X-Forwarded-Proto $scheme;
|
|
}
|
|
location ^~ /o/ {
|
|
proxy_pass http://eusrr_backend;
|
|
proxy_set_header X-Forwarded-Proto $scheme;
|
|
}
|
|
location /media/ {
|
|
alias /srv/eusrr/backend/media/;
|
|
}
|
|
location / {
|
|
proxy_pass http://eusrr_frontend;
|
|
}
|
|
'''
|
|
NGINX = (
|
|
'server {\n listen 80;\n server_name 172.11.0.11;\n'+LOCATIONS+'}\n'
|
|
'server {\n listen 80;\n server_name corp.robotail.local corp.robotail.pro 127.0.0.1;\n'
|
|
' return 301 https://$host$request_uri;\n}\n'
|
|
'server {\n listen 443 ssl;\n server_name corp.robotail.pro;\n'
|
|
' ssl_certificate /etc/ssl/localcerts/corp.robotail.pro.cert.pem;\n'+LOCATIONS+'}\n'
|
|
)
|
|
|
|
|
|
@unittest.skipUnless(os.name == 'posix' and getattr(os, 'geteuid', lambda: -1)() == 0,
|
|
'Disposable root Linux container required')
|
|
class ProxySecurityTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory(prefix='proxy-security-test-')
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
script = SOURCE.read_text()
|
|
for prefix in ('/srv/', '/var/', '/run/', '/etc/'):
|
|
script = script.replace(prefix, str(self.root)+prefix)
|
|
spec = importlib.util.spec_from_loader('proxy_security', loader=None)
|
|
self.module = importlib.util.module_from_spec(spec)
|
|
exec(compile(script, str(SOURCE), 'exec'), self.module.__dict__)
|
|
for directory in ('etc/nginx/sites-available', 'srv/eusrr/backend',
|
|
'var/lib/eusrr-deploy', 'run/lock'):
|
|
(self.root/directory).mkdir(parents=True, exist_ok=True)
|
|
self.nginx = self.root/'etc/nginx/sites-available/eusrr'
|
|
self.nginx_source = NGINX.replace('/srv/', str(self.root)+'/srv/').replace('/etc/', str(self.root)+'/etc/')
|
|
self.nginx.write_text(self.nginx_source)
|
|
self.env = self.root/'srv/eusrr/backend/.env'
|
|
self.env_source = 'DJANGO_SECRET_KEY=dummy-unchanged-key\n'
|
|
self.env.write_text(self.env_source)
|
|
self.env.chmod(0o640)
|
|
|
|
def run_installer(self, flags=(), fail_test=False):
|
|
calls = []
|
|
|
|
def run(args, **kwargs):
|
|
calls.append(args)
|
|
if fail_test and len(calls) == 1:
|
|
raise subprocess.CalledProcessError(1, args)
|
|
return subprocess.CompletedProcess(args, 0)
|
|
|
|
with patch.object(self.module.sys, 'argv', ['installer', *flags]), \
|
|
patch.object(self.module.subprocess, 'run', side_effect=run), \
|
|
contextlib.redirect_stdout(io.StringIO()):
|
|
self.module.main()
|
|
return calls
|
|
|
|
def test_safe_proxy_and_media_rules_in_both_application_vhosts(self):
|
|
self.run_installer(('--protect-media',))
|
|
content = self.nginx.read_text()
|
|
self.assertIn('allow 172.11.1.254;', content)
|
|
self.assertIn('allow 127.0.0.1;', content)
|
|
self.assertIn('deny all;', content)
|
|
self.assertEqual(content.count('location ^~ /_protected_documents/ {'), 2)
|
|
self.assertEqual(content.count('location ^~ /_protected_chat_attachments/ {'), 2)
|
|
self.assertEqual(content.count(' internal;'), 4)
|
|
for prefix in ('/media/documents/', '/smedia/documents/', '/media/task_attachments/',
|
|
'/smedia/task_attachments/', '/media/procurement/item_attachments/',
|
|
'/media/requests/attachments/', '/smedia/requests/attachments/',
|
|
'/media/chat_attachments/', '/smedia/chat_attachments/'):
|
|
self.assertEqual(content.count(f'location ^~ {prefix} {{ return 404; }}'), 2)
|
|
self.assertNotIn('X-Forwarded-Proto $scheme', content)
|
|
self.assertEqual(content.count(' access_log '), 3)
|
|
safe_format = next(line for line in content.splitlines() if line.startswith('log_format'))
|
|
self.assertNotIn('$request ', safe_format)
|
|
self.assertNotIn('$request_uri', safe_format)
|
|
self.assertNotIn('$http_referer', safe_format)
|
|
self.assertIn('DJANGO_SECRET_KEY=dummy-unchanged-key', self.env.read_text())
|
|
self.assertIn('CHAT_ATTACHMENTS_USE_X_ACCEL_REDIRECT=true', self.env.read_text())
|
|
self.assertEqual(self.env.stat().st_mode & 0o777, 0o640)
|
|
|
|
def test_http_only_removes_only_verified_tls_vhost(self):
|
|
self.run_installer(('--http-only', '--protect-media'))
|
|
content = self.nginx.read_text()
|
|
self.assertNotIn('listen 443 ssl', content)
|
|
self.assertEqual(content.count('location ^~ /_protected_documents/ {'), 1)
|
|
self.assertEqual(content.count('proxy_pass http://eusrr_frontend;'), 1)
|
|
|
|
def test_installer_is_idempotent(self):
|
|
self.run_installer(('--protect-media',))
|
|
source, env = self.nginx.read_text(), self.env.read_text()
|
|
self.run_installer(('--protect-media',))
|
|
self.assertEqual(self.nginx.read_text(), source)
|
|
self.assertEqual(self.env.read_text(), env)
|
|
|
|
def test_failed_nginx_validation_restores_config_and_environment(self):
|
|
with self.assertRaises(subprocess.CalledProcessError):
|
|
self.run_installer(('--protect-media',), fail_test=True)
|
|
self.assertEqual(self.nginx.read_text(), self.nginx_source)
|
|
self.assertEqual(self.env.read_text(), self.env_source)
|
|
|
|
def test_unexpected_tls_vhost_aborts_before_any_write(self):
|
|
self.nginx_source = self.nginx_source.replace('corp.robotail.pro.cert.pem', 'other.cert.pem')
|
|
self.nginx.write_text(self.nginx_source)
|
|
with self.assertRaisesRegex(SystemExit, 'Unexpected TLS vhost'):
|
|
self.run_installer(('--http-only',))
|
|
self.assertEqual(self.nginx.read_text(), self.nginx_source)
|
|
self.assertEqual(self.env.read_text(), self.env_source)
|
|
|
|
def test_parent_symlink_is_rejected(self):
|
|
alias = self.root/'alias'
|
|
alias.symlink_to(self.env.parent, target_is_directory=True)
|
|
with self.assertRaises(OSError):
|
|
self.module.read_file(alias/'.env')
|
|
|
|
def test_concurrent_env_change_is_preserved_when_nginx_rolls_back(self):
|
|
original = self.module.atomic_update
|
|
replaced = False
|
|
|
|
def change_env(path, content, expected_source=None):
|
|
nonlocal replaced
|
|
if path == self.env and not replaced:
|
|
replaced = True
|
|
self.env.write_text('concurrent-admin-change\n')
|
|
return original(path, content, expected_source=expected_source)
|
|
|
|
with patch.object(self.module, 'atomic_update', side_effect=change_env):
|
|
with self.assertRaisesRegex(RuntimeError, 'changed concurrently'):
|
|
self.run_installer()
|
|
self.assertEqual(self.nginx.read_text(), self.nginx_source)
|
|
self.assertEqual(self.env.read_text(), 'concurrent-admin-change\n')
|
|
|
|
def test_staging_swap_does_not_modify_symlink_victim(self):
|
|
victim = self.root/'victim'
|
|
victim.write_text('untouched')
|
|
victim.chmod(0o644)
|
|
original = os.fchown
|
|
|
|
def swap(descriptor, uid, gid):
|
|
stage, = self.env.parent.glob('.corp-security-*')
|
|
stage.unlink()
|
|
stage.symlink_to(victim)
|
|
original(descriptor, uid, gid)
|
|
|
|
with patch.object(self.module.os, 'fchown', side_effect=swap):
|
|
with self.assertRaisesRegex(RuntimeError, 'Staging file was replaced'):
|
|
self.module.atomic_update(self.env, 'new')
|
|
self.assertEqual(victim.read_text(), 'untouched')
|
|
self.assertEqual(victim.stat().st_mode & 0o777, 0o644)
|
|
self.assertEqual(self.env.read_text(), self.env_source)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|