mteehan
835326311b
Replace the legacy top-level management key with a unified paths-based model. Each domain now contains a paths map where each entry defines its own backend, auth, headers, and flags (is_management, is_websocket). - Add _migrate_config() to auto-migrate legacy formats on first load - Remove set_management_proxy() and POST_NGINX_MANAGEMENT endpoint - Update server_block.conf template to iterate paths with per-location auth - Update daemon handler, API blueprint, state collector, and install script - Add server config generation tests for paths, WebSocket, auth inheritance - Update frontend proxy page to display per-path rows with flags
383 lines
13 KiB
Python
383 lines
13 KiB
Python
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from lib import nginx
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_default_config():
|
|
"""Reset the shared mutable DEFAULT_CONFIG before each test."""
|
|
original = nginx.DEFAULT_CONFIG.copy()
|
|
yield
|
|
# Reset the shared "domains" dict that leaks due to shallow copy in _json_load
|
|
nginx.DEFAULT_CONFIG = original
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_data_dir(tmp_path):
|
|
original_config = nginx.CONFIG_FILE
|
|
original_sites = nginx.SITES_DIR
|
|
original_htpasswd = nginx.HTPASSWD_FILE
|
|
original_ssl_snippet = nginx.SSL_SNIPPET
|
|
original_include = nginx.INCLUDE_FILE
|
|
original_config_dir = nginx.CONFIG_DIR
|
|
original_data_dir = nginx.DATA_DIR
|
|
|
|
nginx.CONFIG_DIR = tmp_path / "nginx"
|
|
nginx.DATA_DIR = tmp_path / "nginx"
|
|
nginx.SITES_DIR = tmp_path / "nginx" / "sites-enabled"
|
|
nginx.CONFIG_FILE = tmp_path / "nginx" / "config.json"
|
|
nginx.HTPASSWD_FILE = tmp_path / "nginx" / ".htpasswd"
|
|
nginx.SSL_SNIPPET = tmp_path / "ssl_snippet.conf"
|
|
nginx.INCLUDE_FILE = tmp_path / "include.conf"
|
|
|
|
nginx.CONFIG_DIR.mkdir(parents=True, exist_ok=True)
|
|
nginx.DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
nginx.SITES_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
yield tmp_path
|
|
|
|
nginx.CONFIG_FILE = original_config
|
|
nginx.SITES_DIR = original_sites
|
|
nginx.HTPASSWD_FILE = original_htpasswd
|
|
nginx.SSL_SNIPPET = original_ssl_snippet
|
|
nginx.INCLUDE_FILE = original_include
|
|
nginx.CONFIG_DIR = original_config_dir
|
|
nginx.DATA_DIR = original_data_dir
|
|
|
|
|
|
class TestGetConfig:
|
|
def test_returns_default_when_no_file(self, temp_data_dir):
|
|
cfg = nginx.get_config()
|
|
assert "domains" in cfg
|
|
assert "ssl" in cfg
|
|
assert cfg["domains"] == {}
|
|
|
|
|
|
class TestSaveConfig:
|
|
def test_saves_and_reloads(self, temp_data_dir):
|
|
cfg = {
|
|
"domains": {"example.com": {"backend": {"host": "localhost", "port": 80}}}
|
|
}
|
|
nginx.save_config(cfg)
|
|
loaded = nginx.get_config()
|
|
assert (
|
|
loaded["domains"]["example.com"]["paths"]["/"]["backend"]["host"]
|
|
== "localhost"
|
|
)
|
|
|
|
|
|
class TestGetDomains:
|
|
def test_empty_domains(self, temp_data_dir):
|
|
result = nginx.get_domains()
|
|
assert result == []
|
|
|
|
def test_returns_domain_list(self, temp_data_dir):
|
|
cfg = {
|
|
"domains": {
|
|
"example.com": {
|
|
"backend": {"host": "localhost", "port": 8080, "proto": "http"},
|
|
"force_ssl": True,
|
|
}
|
|
}
|
|
}
|
|
nginx.save_config(cfg)
|
|
result = nginx.get_domains()
|
|
assert len(result) == 1
|
|
assert result[0]["domain"] == "example.com"
|
|
|
|
|
|
class TestAddDomain:
|
|
@patch("lib.nginx.get_config")
|
|
def test_add_domain(self, mock_get, temp_data_dir):
|
|
mock_get.return_value = {
|
|
"domains": {},
|
|
"management": None,
|
|
"ssl": {"protocols": "TLSv1.2 TLSv1.3"},
|
|
}
|
|
nginx.add_domain("example.com", "10.0.0.5", 8080)
|
|
cfg = nginx.get_config()
|
|
assert "example.com" in cfg["domains"]
|
|
assert (
|
|
cfg["domains"]["example.com"]["paths"]["/"]["backend"]["host"] == "10.0.0.5"
|
|
)
|
|
assert cfg["domains"]["example.com"]["paths"]["/"]["backend"]["port"] == 8080
|
|
|
|
@patch("lib.nginx.get_config")
|
|
def test_duplicate_domain_raises(self, mock_get, temp_data_dir):
|
|
mock_get.return_value = {
|
|
"domains": {
|
|
"example.com": {"backend": {"host": "x", "port": 80, "proto": "http"}}
|
|
},
|
|
"management": None,
|
|
"ssl": {"protocols": "TLSv1.2 TLSv1.3"},
|
|
}
|
|
with pytest.raises(ValueError):
|
|
nginx.add_domain("example.com", "10.0.0.5", 8080)
|
|
|
|
|
|
class TestRemoveDomain:
|
|
@patch("lib.nginx.get_config")
|
|
def test_remove_existing_domain(self, mock_get, temp_data_dir):
|
|
mock_get.return_value = {
|
|
"domains": {
|
|
"example.com": {"backend": {"host": "x", "port": 80, "proto": "http"}}
|
|
},
|
|
"management": None,
|
|
"ssl": {"protocols": "TLSv1.2 TLSv1.3"},
|
|
}
|
|
nginx.remove_domain("example.com")
|
|
cfg = nginx.get_config()
|
|
assert "example.com" not in cfg["domains"]
|
|
|
|
def test_remove_nonexistent_domain(self, temp_data_dir):
|
|
nginx.remove_domain("nonexistent.com")
|
|
cfg = nginx.get_config()
|
|
assert "nonexistent.com" not in cfg["domains"]
|
|
|
|
|
|
class TestUpdateDomain:
|
|
@patch("lib.nginx.get_config")
|
|
def test_update_existing_domain(self, mock_get, temp_data_dir):
|
|
entry = {
|
|
"backend": {"host": "10.0.0.5", "port": 8080, "proto": "http"},
|
|
"force_ssl": True,
|
|
}
|
|
mock_get.return_value = {
|
|
"domains": {"example.com": entry},
|
|
"management": None,
|
|
"ssl": {"protocols": "TLSv1.2 TLSv1.3"},
|
|
}
|
|
nginx.update_domain("example.com", force_ssl=False)
|
|
cfg = nginx.get_config()
|
|
assert cfg["domains"]["example.com"]["force_ssl"] is False
|
|
|
|
def test_update_nonexistent_raises(self, temp_data_dir):
|
|
with pytest.raises(KeyError):
|
|
nginx.update_domain("nonexistent.com", force_ssl=False)
|
|
|
|
|
|
class TestWriteSite:
|
|
def test_write_creates_file(self, temp_data_dir):
|
|
nginx.write_site("example.com", "server { listen 443; }")
|
|
path = nginx.SITES_DIR / "example.com.conf"
|
|
assert path.exists()
|
|
content = Path(path).read_text()
|
|
assert "server { listen 443; }" in content
|
|
|
|
|
|
class TestGenerateServerConf:
|
|
def test_simple_root_path(self, temp_data_dir):
|
|
cfg = {
|
|
"domain": "example.com",
|
|
"paths": {
|
|
"/": {
|
|
"backend": {"host": "10.0.0.1", "port": 80, "proto": "http"},
|
|
"headers": {"X-Custom": "value"},
|
|
}
|
|
},
|
|
"force_ssl": True,
|
|
"cert": "acme",
|
|
}
|
|
out = nginx.generate_server_conf(cfg)
|
|
assert "location /" in out
|
|
assert "proxy_pass http://10.0.0.1:80;" in out
|
|
assert "proxy_set_header X-Custom value;" in out
|
|
assert "add_header X-Content-Type-Options" in out
|
|
|
|
def test_multiple_paths(self, temp_data_dir):
|
|
cfg = {
|
|
"domain": "app.example.com",
|
|
"paths": {
|
|
"/": {
|
|
"backend": {"host": "10.0.0.1", "port": 80, "proto": "http"},
|
|
"headers": {},
|
|
},
|
|
"/api": {
|
|
"backend": {"host": "10.0.0.2", "port": 8080, "proto": "http"},
|
|
},
|
|
},
|
|
"force_ssl": True,
|
|
"cert": "acme",
|
|
}
|
|
out = nginx.generate_server_conf(cfg)
|
|
assert "proxy_pass http://10.0.0.1:80;" in out
|
|
assert "proxy_pass http://10.0.0.2:8080;" in out
|
|
assert "location /api" in out
|
|
|
|
def test_management_path(self, temp_data_dir):
|
|
cfg = {
|
|
"domain": "mgmt.example.com",
|
|
"paths": {
|
|
"/": {
|
|
"backend": {"host": "127.0.0.1", "port": 9090, "proto": "http"},
|
|
"is_management": True,
|
|
}
|
|
},
|
|
"force_ssl": True,
|
|
"cert": "acme",
|
|
"auth": {"user": "admin", "htpasswd": "/path/.htpasswd"},
|
|
}
|
|
out = nginx.generate_server_conf(cfg)
|
|
assert "proxy_pass http://127.0.0.1:9090;" in out
|
|
assert "add_header X-Content-Type-Options" not in out
|
|
assert "wall_mgmt_access.log" in out
|
|
|
|
def test_websocket_path(self, temp_data_dir):
|
|
cfg = {
|
|
"domain": "mgmt.example.com",
|
|
"paths": {
|
|
"/": {
|
|
"backend": {"host": "127.0.0.1", "port": 9090, "proto": "http"},
|
|
},
|
|
"/ws": {
|
|
"backend": {"host": "127.0.0.1", "port": 9091, "proto": "http"},
|
|
"is_websocket": True,
|
|
},
|
|
},
|
|
"force_ssl": True,
|
|
"cert": "acme",
|
|
}
|
|
out = nginx.generate_server_conf(cfg)
|
|
assert "proxy_pass http://127.0.0.1:9091;" in out
|
|
assert "proxy_set_header Upgrade" in out
|
|
assert "proxy_read_timeout 86400s;" in out
|
|
|
|
def test_auth_inheritance(self, temp_data_dir):
|
|
cfg = {
|
|
"domain": "app.example.com",
|
|
"paths": {
|
|
"/": {
|
|
"backend": {"host": "10.0.0.1", "port": 80, "proto": "http"},
|
|
"headers": {},
|
|
},
|
|
"/api": {
|
|
"backend": {"host": "10.0.0.2", "port": 8080, "proto": "http"},
|
|
"auth": None,
|
|
},
|
|
"/admin": {
|
|
"backend": {"host": "10.0.0.3", "port": 9000, "proto": "http"},
|
|
"auth": {"user": "admin", "htpasswd": "/other/.htpasswd"},
|
|
},
|
|
},
|
|
"force_ssl": True,
|
|
"cert": "acme",
|
|
"auth": {"user": "admin", "htpasswd": "/path/.htpasswd"},
|
|
}
|
|
out = nginx.generate_server_conf(cfg)
|
|
assert "auth_basic_user_file /path/.htpasswd;" in out
|
|
lines = out.split("\n")
|
|
api_idx = next(i for i, line in enumerate(lines) if "location /api" in line)
|
|
admin_idx = next(i for i, line in enumerate(lines) if "location /admin" in line)
|
|
# /api should have auth_basic off
|
|
assert "auth_basic off;" in "\n".join(lines[api_idx : api_idx + 5])
|
|
# /admin should have path-level auth override
|
|
assert "auth_basic_user_file /other/.htpasswd;" in "\n".join(
|
|
lines[admin_idx : admin_idx + 5]
|
|
)
|
|
|
|
|
|
class TestWriteAllSites:
|
|
@patch("lib.nginx.get_config")
|
|
def test_writes_all_domains(self, mock_get, temp_data_dir):
|
|
mock_get.return_value = {
|
|
"domains": {
|
|
"a.com": {
|
|
"paths": {
|
|
"/": {
|
|
"backend": {
|
|
"host": "10.0.0.1",
|
|
"port": 80,
|
|
"proto": "http",
|
|
},
|
|
"headers": {},
|
|
}
|
|
},
|
|
"force_ssl": True,
|
|
},
|
|
"b.com": {
|
|
"paths": {
|
|
"/": {
|
|
"backend": {
|
|
"host": "10.0.0.2",
|
|
"port": 80,
|
|
"proto": "http",
|
|
},
|
|
"headers": {},
|
|
}
|
|
},
|
|
"force_ssl": True,
|
|
},
|
|
},
|
|
"ssl": {"protocols": "TLSv1.2 TLSv1.3"},
|
|
}
|
|
nginx.write_all_sites()
|
|
assert (nginx.SITES_DIR / "a.com.conf").exists()
|
|
assert (nginx.SITES_DIR / "b.com.conf").exists()
|
|
|
|
@patch("lib.nginx.get_config")
|
|
def test_removes_old_sites(self, mock_get, temp_data_dir):
|
|
# Pre-create an old site
|
|
nginx.write_site("old.com", "server {}")
|
|
assert (nginx.SITES_DIR / "old.com.conf").exists()
|
|
|
|
mock_get.return_value = {
|
|
"domains": {},
|
|
"management": None,
|
|
"ssl": {"protocols": "TLSv1.2 TLSv1.3"},
|
|
}
|
|
nginx.write_all_sites()
|
|
assert not (nginx.SITES_DIR / "old.com.conf").exists()
|
|
|
|
|
|
class TestTestConfig:
|
|
@patch("lib.nginx.subprocess.run")
|
|
def test_passes(self, mock_run, temp_data_dir):
|
|
mock_run.return_value = MagicMock(
|
|
returncode=0, stdout="", stderr="test passed\n"
|
|
)
|
|
ok, _msg = nginx.test_config()
|
|
assert ok is True
|
|
|
|
@patch("lib.nginx.subprocess.run")
|
|
def test_fails(self, mock_run, temp_data_dir):
|
|
mock_run.return_value = MagicMock(
|
|
returncode=1, stdout="", stderr="nginx: configuration test failed\n"
|
|
)
|
|
ok, _msg = nginx.test_config()
|
|
assert ok is False
|
|
|
|
|
|
class TestWriteHtpasswd:
|
|
@patch("lib.nginx._hash_password")
|
|
def test_creates_file(self, mock_hash, temp_data_dir):
|
|
mock_hash.return_value = "$apr1$hash"
|
|
nginx.write_htpasswd("admin", "secret")
|
|
assert nginx.HTPASSWD_FILE.exists()
|
|
content = nginx.HTPASSWD_FILE.read_text()
|
|
assert "admin:" in content
|
|
|
|
@patch("lib.nginx._hash_password")
|
|
def test_replaces_existing_user(self, mock_hash, temp_data_dir):
|
|
mock_hash.return_value = "$apr1$hash1"
|
|
nginx.write_htpasswd("admin", "old")
|
|
mock_hash.return_value = "$apr1$hash2"
|
|
nginx.write_htpasswd("admin", "new")
|
|
lines = [
|
|
line
|
|
for line in nginx.HTPASSWD_FILE.read_text().strip().splitlines()
|
|
if line
|
|
]
|
|
assert len([line for line in lines if line.startswith("admin:")]) == 1
|
|
|
|
|
|
class TestHashPasswordFallback:
|
|
@patch("lib.nginx._hash_password")
|
|
def test_hash_returns_string(self, mock_hash, temp_data_dir):
|
|
mock_hash.return_value = "$apr1$hash"
|
|
result = nginx._hash_password("test")
|
|
assert isinstance(result, str)
|
|
assert len(result) > 0
|