123 lines
5.4 KiB
Python
123 lines
5.4 KiB
Python
import socket
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from starlette.requests import Request
|
|
|
|
from backend.app.core.network_security import UnsafeConnectorUrl, validate_connector_url
|
|
|
|
|
|
def test_connector_url_allows_private_lan_hosts(monkeypatch):
|
|
monkeypatch.setattr(socket, "getaddrinfo", lambda *_args, **_kwargs: (_ for _ in ()).throw(socket.gaierror()))
|
|
assert validate_connector_url("http://198.18.1.10:9000/") == "http://198.18.1.10:9000"
|
|
assert validate_connector_url("https://sonarr.internal/api") == "https://sonarr.internal/api"
|
|
|
|
|
|
def test_connector_url_blocks_unsafe_targets():
|
|
blocked = [
|
|
"file:///etc/passwd",
|
|
"http://localhost:8080",
|
|
"http://127.0.0.1:8080",
|
|
"http://169.254.169.254/latest/meta-data",
|
|
"http://user:pass@host.local",
|
|
"http://0.0.0.0:80",
|
|
]
|
|
for value in blocked:
|
|
try:
|
|
validate_connector_url(value)
|
|
except UnsafeConnectorUrl:
|
|
continue
|
|
raise AssertionError(f"unsafe connector URL accepted: {value}")
|
|
|
|
|
|
def test_connector_url_blocks_dns_names_resolving_to_loopback(monkeypatch):
|
|
monkeypatch.setattr(socket, "getaddrinfo", lambda *_args, **_kwargs: [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", 80))])
|
|
with pytest.raises(UnsafeConnectorUrl):
|
|
validate_connector_url("http://connector.internal")
|
|
|
|
|
|
def _request(headers: list[tuple[bytes, bytes]] | None = None, client_host: str = "198.18.10.20") -> Request:
|
|
return Request({"type": "http", "method": "POST", "path": "/api/actions/plan", "headers": headers or [], "client": (client_host, 12345)})
|
|
|
|
|
|
def test_operator_boundary_protects_mutations_and_accepts_valid_token():
|
|
from backend.app.core import security
|
|
from backend.app.core.config import settings
|
|
|
|
previous = (settings.operator_auth_required, settings.operator_token, security.settings.operator_auth_required, security.settings.operator_token)
|
|
settings.operator_auth_required = True
|
|
settings.operator_token = "test-operator-boundary-token"
|
|
security.settings.operator_auth_required = True
|
|
security.settings.operator_token = "test-operator-boundary-token"
|
|
try:
|
|
with pytest.raises(HTTPException) as missing:
|
|
security.require_operator_request(_request())
|
|
assert missing.value.status_code == 401
|
|
request = _request([(b"x-appops-operator-token", b"test-operator-boundary-token")])
|
|
assert security.require_operator_request(request) == "operator"
|
|
finally:
|
|
settings.operator_auth_required, settings.operator_token, security.settings.operator_auth_required, security.settings.operator_token = previous
|
|
|
|
|
|
def test_disabled_operator_boundary_ignores_stale_configured_token():
|
|
from backend.app.core import security
|
|
from backend.app.core.config import settings
|
|
|
|
previous = (settings.operator_auth_required, settings.operator_token)
|
|
settings.operator_auth_required = False
|
|
settings.operator_token = "stale-operator-token-that-no-longer-unlocks"
|
|
try:
|
|
assert security.require_operator_request(_request()) == "private_unlocked"
|
|
assert security.operator_session_status(_request()) == {
|
|
"authorized": True,
|
|
"mode": "private",
|
|
"identity": None,
|
|
}
|
|
finally:
|
|
settings.operator_auth_required, settings.operator_token = previous
|
|
|
|
|
|
def test_operator_boundary_accepts_allowed_authentik_account_only_from_trusted_proxy():
|
|
from backend.app.core import security
|
|
from backend.app.core.config import settings
|
|
|
|
previous = (
|
|
settings.operator_auth_required,
|
|
settings.operator_token,
|
|
settings.operator_proxy_auth_enabled,
|
|
settings.operator_proxy_trusted_networks,
|
|
settings.operator_proxy_users,
|
|
settings.operator_proxy_groups,
|
|
)
|
|
settings.operator_auth_required = True
|
|
settings.operator_token = "fallback-operator-token-value"
|
|
settings.operator_proxy_auth_enabled = True
|
|
settings.operator_proxy_trusted_networks = ["203.0.113.0/16"]
|
|
settings.operator_proxy_users = ["example-host"]
|
|
settings.operator_proxy_groups = ["appops-operators"]
|
|
headers = [(b"x-authentik-username", b"example-host"), (b"x-authentik-groups", b"users|appops-operators")]
|
|
try:
|
|
assert security.require_operator_request(_request(headers, "203.0.113.8")) == "authentik:example-host"
|
|
assert security.operator_session_status(_request(headers, "203.0.113.8")) == {
|
|
"authorized": True,
|
|
"mode": "account",
|
|
"identity": "example-host",
|
|
}
|
|
assert security.require_operator_request(_request(headers + [(b"x-appops-operator-token", b"stale-token")], "203.0.113.8")) == "authentik:example-host"
|
|
with pytest.raises(HTTPException) as spoofed:
|
|
security.require_operator_request(_request(headers, "198.18.10.20"))
|
|
assert spoofed.value.status_code == 401
|
|
denied_headers = [(b"x-authentik-username", b"someone-else"), (b"x-authentik-groups", b"users")]
|
|
with pytest.raises(HTTPException) as denied:
|
|
security.require_operator_request(_request(denied_headers, "203.0.113.8"))
|
|
assert denied.value.status_code == 401
|
|
finally:
|
|
(
|
|
settings.operator_auth_required,
|
|
settings.operator_token,
|
|
settings.operator_proxy_auth_enabled,
|
|
settings.operator_proxy_trusted_networks,
|
|
settings.operator_proxy_users,
|
|
settings.operator_proxy_groups,
|
|
) = previous
|