2ea038504d
Some checks failed
CD Pipeline / select-latest-carrier (push) Successful in 45s
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m19s
CD Pipeline / revalidate-deploy-carrier (push) Successful in 43s
CD Pipeline / revalidate-post-deploy-carrier (push) Has been cancelled
CD Pipeline / post-deploy-checks (push) Has been cancelled
CD Pipeline / build-and-deploy (push) Has been cancelled
279 lines
11 KiB
Python
279 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import stat
|
|
import threading
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[3]
|
|
TARGET = ROOT / "agent99-signoz-credential-target.py"
|
|
PROVISIONER = ROOT / "agent99-signoz-credential-provisioner.ps1"
|
|
|
|
|
|
def load_target():
|
|
spec = importlib.util.spec_from_file_location("signoz_credential_target", TARGET)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class SignozState:
|
|
setup_completed = False
|
|
fail_dashboard = False
|
|
fail_session_cleanup = False
|
|
lose_register_response_once = False
|
|
deny_pat_admin_readback = False
|
|
requests: list[tuple[str, str, dict[str, Any] | None]]
|
|
|
|
def __init__(self) -> None:
|
|
self.setup_completed = False
|
|
self.fail_dashboard = False
|
|
self.fail_session_cleanup = False
|
|
self.lose_register_response_once = False
|
|
self.deny_pat_admin_readback = False
|
|
self.requests = []
|
|
|
|
|
|
def start_server(state: SignozState) -> tuple[ThreadingHTTPServer, str]:
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self, *_args: object) -> None:
|
|
return
|
|
|
|
def _send(self, status: int, value: dict[str, Any]) -> None:
|
|
raw = json.dumps(value).encode()
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(raw)))
|
|
self.end_headers()
|
|
self.wfile.write(raw)
|
|
|
|
def _no_content(self) -> None:
|
|
self.send_response(204)
|
|
self.end_headers()
|
|
|
|
def _body(self) -> dict[str, Any] | None:
|
|
length = int(self.headers.get("Content-Length", "0"))
|
|
return json.loads(self.rfile.read(length)) if length else None
|
|
|
|
def do_GET(self) -> None: # noqa: N802
|
|
state.requests.append(("GET", self.path, None))
|
|
if self.path == "/api/v1/version":
|
|
self._send(
|
|
200,
|
|
{"version": "v0.113.0", "ee": "Y", "setupCompleted": state.setup_completed},
|
|
)
|
|
elif self.path == "/api/v1/user/me":
|
|
self._send(200, {"status": "success", "data": {"role": "ADMIN", "orgId": "org-1"}})
|
|
elif self.path.startswith("/api/v2/sessions/context?"):
|
|
self._send(
|
|
200,
|
|
{
|
|
"status": "success",
|
|
"data": {"exists": True, "orgs": [{"id": "org-1"}]},
|
|
},
|
|
)
|
|
elif self.path == "/api/v1/dashboards" and state.fail_dashboard:
|
|
self._send(401, {"status": "error"})
|
|
elif self.path == "/api/v1/dashboards":
|
|
assert self.headers.get("SIGNOZ-API-KEY") == "opaque-pat-value"
|
|
self._send(200, {"status": "success", "data": []})
|
|
elif self.path == "/api/v1/roles":
|
|
self._send(200, {"status": "success", "data": [{"name": "ADMIN"}]})
|
|
elif self.path == "/api/v1/pats" and state.deny_pat_admin_readback:
|
|
self._send(403, {"status": "error"})
|
|
elif self.path == "/api/v1/pats":
|
|
self._send(200, {"status": "success", "data": [{"id": "pat-1"}]})
|
|
else:
|
|
self._send(404, {"status": "error"})
|
|
|
|
def do_POST(self) -> None: # noqa: N802
|
|
body = self._body()
|
|
state.requests.append(("POST", self.path, body))
|
|
if self.path == "/api/v1/register":
|
|
state.setup_completed = True
|
|
if state.lose_register_response_once:
|
|
state.lose_register_response_once = False
|
|
self.connection.shutdown(2)
|
|
self.connection.close()
|
|
return
|
|
self._send(200, {"status": "success", "data": {"orgId": "org-1"}})
|
|
elif self.path == "/api/v2/sessions/email_password":
|
|
self._send(
|
|
200,
|
|
{"status": "success", "data": {"accessToken": "session-jwt", "refreshToken": "refresh"}},
|
|
)
|
|
elif self.path == "/api/v1/pats":
|
|
self._send(
|
|
201,
|
|
{"status": "success", "data": {"id": "pat-1", "token": "opaque-pat-value"}},
|
|
)
|
|
else:
|
|
self._send(404, {"status": "error"})
|
|
|
|
def do_DELETE(self) -> None: # noqa: N802
|
|
state.requests.append(("DELETE", self.path, None))
|
|
if self.path == "/api/v2/sessions" and state.fail_session_cleanup:
|
|
self._send(500, {"status": "error"})
|
|
else:
|
|
self._no_content()
|
|
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
host, port = server.server_address
|
|
return server, f"http://{host}:{port}"
|
|
|
|
|
|
def configure(module: Any, tmp_path: Path, base_url: str) -> Path:
|
|
credential = tmp_path / "secrets" / "signoz-metadata-api-key"
|
|
module.BASE_URL = base_url
|
|
module.CREDENTIAL_PATH = credential
|
|
module.RECOVERY_PASSWORD_PATH = tmp_path / "secrets" / "bootstrap.recovery"
|
|
module.EXPECTED_UID = os.geteuid()
|
|
module.os.chown = lambda *_args, **_kwargs: None
|
|
return credential
|
|
|
|
|
|
def test_agent99_entrypoint_is_fixed_and_never_transports_secret() -> None:
|
|
source = PROVISIONER.read_text(encoding="utf-8")
|
|
target = TARGET.read_text(encoding="utf-8")
|
|
|
|
assert '[ValidateSet("Check", "Apply", "Verify")]' in source
|
|
assert '$TargetHost = "192.168.0.110"' in source
|
|
assert '$ExpectedWorkItemId = "P0-OBS-002"' in source
|
|
assert '$ExpectedRuntimeFileCount = 20' in source
|
|
assert '"sudo -n /usr/bin/python3 - --mode $($Mode.ToLowerInvariant())"' in source
|
|
assert "$process.Refresh()" in source
|
|
assert 'if (-not $process.HasExited) { throw "target_exit_state_unavailable" }' in source
|
|
assert "$exitCode = [int]$process.ExitCode" in source
|
|
assert '$exitCode -ne 0 -or [string]$result.terminal -eq "failed"' in source
|
|
assert "-RedirectStandardInput $TargetProgram" in source
|
|
assert "secret_in_transport = $false" in source
|
|
assert "Invoke-Expression" not in source
|
|
assert "-EncodedCommand" not in source
|
|
assert '/etc/awoooi/secrets/signoz-metadata-api-key' in target
|
|
assert "os.O_EXCL" in target and "os.O_NOFOLLOW" in target
|
|
assert '"role": "ADMIN"' in target
|
|
assert '"expiresInDays": 30' in target
|
|
assert "SIGNOZ-API-KEY" in target
|
|
assert "status != 204" in target
|
|
assert "token_urlsafe" in target
|
|
assert "setup_already_completed_without_bootstrap_recovery" in target
|
|
assert "bootstrap_recovery_removed" in target
|
|
assert "print(password" not in target
|
|
assert "print(token" not in target
|
|
|
|
|
|
def test_apply_bootstraps_admin_persists_only_pat_and_verifies(tmp_path: Path) -> None:
|
|
module = load_target()
|
|
state = SignozState()
|
|
server, base_url = start_server(state)
|
|
credential = configure(module, tmp_path, base_url)
|
|
try:
|
|
result = module.apply()
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
|
|
assert result["terminal"] == "applied_verified"
|
|
assert result["bootstrap_is_permanent"] is True
|
|
assert credential.read_text() == "opaque-pat-value"
|
|
assert stat.S_IMODE(credential.stat().st_mode) == 0o400
|
|
assert not module.RECOVERY_PASSWORD_PATH.exists()
|
|
serialized = json.dumps(result)
|
|
assert "opaque-pat-value" not in serialized
|
|
assert "session-jwt" not in serialized
|
|
assert any(method == "DELETE" and path == "/api/v2/sessions" for method, path, _ in state.requests)
|
|
|
|
|
|
def test_apply_failure_revokes_pat_and_removes_credential(tmp_path: Path) -> None:
|
|
module = load_target()
|
|
state = SignozState()
|
|
state.fail_dashboard = True
|
|
server, base_url = start_server(state)
|
|
credential = configure(module, tmp_path, base_url)
|
|
try:
|
|
with pytest.raises(module.ProvisionError, match="http_get_dashboards_401"):
|
|
module.apply()
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
|
|
assert not credential.exists()
|
|
assert module.RECOVERY_PASSWORD_PATH.exists()
|
|
assert ("DELETE", "/api/v1/pats/pat-1", None) in state.requests
|
|
assert ("DELETE", "/api/v2/sessions", None) in state.requests
|
|
|
|
|
|
def test_lost_registration_response_recovers_with_root_only_secret(tmp_path: Path) -> None:
|
|
module = load_target()
|
|
state = SignozState()
|
|
state.lose_register_response_once = True
|
|
server, base_url = start_server(state)
|
|
credential = configure(module, tmp_path, base_url)
|
|
try:
|
|
result = module.apply()
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
|
|
assert result["terminal"] == "applied_verified"
|
|
assert credential.exists()
|
|
assert not module.RECOVERY_PASSWORD_PATH.exists()
|
|
assert any(path.startswith("/api/v2/sessions/context?") for _, path, _ in state.requests)
|
|
|
|
|
|
def test_session_cleanup_failure_revokes_pat_and_preserves_recovery(tmp_path: Path) -> None:
|
|
module = load_target()
|
|
state = SignozState()
|
|
state.fail_session_cleanup = True
|
|
server, base_url = start_server(state)
|
|
credential = configure(module, tmp_path, base_url)
|
|
try:
|
|
with pytest.raises(module.ProvisionError, match="session_cleanup_unverified"):
|
|
module.apply()
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
|
|
assert not credential.exists()
|
|
assert module.RECOVERY_PASSWORD_PATH.exists()
|
|
assert ("DELETE", "/api/v1/pats/pat-1", None) in state.requests
|
|
|
|
|
|
def test_admin_pat_readback_failure_revokes_before_persist(tmp_path: Path) -> None:
|
|
module = load_target()
|
|
state = SignozState()
|
|
state.deny_pat_admin_readback = True
|
|
server, base_url = start_server(state)
|
|
credential = configure(module, tmp_path, base_url)
|
|
try:
|
|
with pytest.raises(module.ProvisionError, match="http_get_pats_403"):
|
|
module.apply()
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
|
|
assert not credential.exists()
|
|
assert module.RECOVERY_PASSWORD_PATH.exists()
|
|
assert ("DELETE", "/api/v1/pats/pat-1", None) in state.requests
|
|
|
|
|
|
def test_verify_rejects_unsafe_credential_metadata(tmp_path: Path) -> None:
|
|
module = load_target()
|
|
credential = tmp_path / "credential"
|
|
credential.write_text("opaque-pat-value")
|
|
credential.chmod(0o644)
|
|
module.CREDENTIAL_PATH = credential
|
|
module.EXPECTED_UID = os.geteuid()
|
|
with pytest.raises(module.ProvisionError, match="credential_mode_invalid"):
|
|
module.credential_state(read_value=True)
|