141 lines
6.5 KiB
Python
141 lines
6.5 KiB
Python
"""Runtime readiness is independent of storage for enrolling a new target."""
|
|
from __future__ import annotations
|
|
|
|
import http.client
|
|
import io
|
|
import json
|
|
import threading
|
|
from email.message import Message
|
|
|
|
import pytest
|
|
from test_deploy_broker import BROKER_ROOT, _load_module
|
|
|
|
|
|
@pytest.fixture
|
|
def controller(monkeypatch, tmp_path):
|
|
module = _load_module("appops_runtime_capabilities_test", BROKER_ROOT / "status_server.py")
|
|
repository_root = tmp_path / "repos"
|
|
repository_root.mkdir()
|
|
secret_path = tmp_path / "bootstrap"
|
|
secret = b"runtime-capabilities-test-" + b"x" * 48
|
|
secret_path.write_bytes(secret)
|
|
secret_path.chmod(0o600)
|
|
monkeypatch.setattr(module, "APPOPS_ENROLLMENT_BOOTSTRAP_SECRET_FILE", secret_path)
|
|
monkeypatch.setattr(module, "APPOPS_NONCE_ROOT", tmp_path / "nonces")
|
|
monkeypatch.setattr(module, "REPOSITORY_ROOT", repository_root)
|
|
monkeypatch.setattr(module, "_controller_release", lambda: {"revision": "a" * 40})
|
|
monkeypatch.setattr(module, "_projects_registry_is_safe", lambda: True)
|
|
monkeypatch.setattr(module, "_load_projects", lambda: {"owner/existing": {}})
|
|
monkeypatch.setattr(module, "_refresh_appdata_root_selection", lambda: None)
|
|
monkeypatch.setattr(module, "trusted_appdata_roots_readiness", lambda: [
|
|
{"root": "/srv/example/appdata", "status": "unavailable", "reason": "unsafe_permissions"},
|
|
])
|
|
monkeypatch.setattr(module, "managed_storage_readiness", lambda: {"status": "needs_initialization"})
|
|
server = module.BoundedThreadingHTTPServer(("127.0.0.1", 0), module.Handler)
|
|
serving = threading.Thread(target=server.serve_forever, daemon=True)
|
|
serving.start()
|
|
counter = 0
|
|
|
|
def request(*, authenticated=True, replay_headers=None):
|
|
nonlocal counter
|
|
counter += 1
|
|
body = b"{}"
|
|
path = "/v1/appops/enrollments/capabilities"
|
|
timestamp = str(int(module.time.time()))
|
|
nonce = f"{counter:032x}"
|
|
headers = replay_headers or {
|
|
"Content-Type": "application/json",
|
|
"X-ITWorx-Deployment-Timestamp": timestamp,
|
|
"X-ITWorx-Deployment-Nonce": nonce,
|
|
"X-ITWorx-Deployment-Signature": module._request_signature(
|
|
secret, method="POST", path=path, timestamp=timestamp, nonce=nonce, raw_body=body,
|
|
) if authenticated else "0" * 64,
|
|
}
|
|
client = http.client.HTTPConnection(*server.server_address, timeout=2)
|
|
client.request("POST", path, body=body, headers=headers)
|
|
response = client.getresponse()
|
|
payload = json.loads(response.read())
|
|
client.close()
|
|
assert secret.decode() not in json.dumps(payload)
|
|
return response.status, payload, headers
|
|
|
|
yield module, request
|
|
server.shutdown()
|
|
server.server_close()
|
|
serving.join(timeout=2)
|
|
|
|
|
|
def test_registered_runtime_ready_without_new_enrollment_storage(controller):
|
|
_, request = controller
|
|
status, payload, headers = request()
|
|
assert status == 200
|
|
assert payload["status"] == "ready"
|
|
assert payload["registry"] == {"status": "ready", "registeredProjectCount": 1}
|
|
assert payload["enrollmentReadiness"] == {
|
|
"status": "unavailable", "reason": "trusted_appdata_roots_unavailable",
|
|
}
|
|
for name in ("deploymentApply", "deploymentVerify", "rollback", "exactTargetPolicy", "hmacV2", "replayProtection"):
|
|
assert payload["features"][name] is True
|
|
for name in ("enrollment", "targetRegistration", "runtimeSecretSealing"):
|
|
assert payload["features"][name] is False
|
|
assert request(replay_headers=headers)[0] == 403
|
|
assert request(authenticated=False)[0] == 403
|
|
|
|
|
|
@pytest.mark.parametrize("unavailable", ["registry", "registry_content", "artifact", "repository"])
|
|
def test_runtime_prerequisites_remain_fail_closed(controller, monkeypatch, tmp_path, unavailable):
|
|
module, request = controller
|
|
if unavailable == "registry":
|
|
monkeypatch.setattr(module, "_projects_registry_is_safe", lambda: False)
|
|
elif unavailable == "registry_content":
|
|
def invalid_registry():
|
|
raise ValueError("invalid registry")
|
|
monkeypatch.setattr(module, "_load_projects", invalid_registry)
|
|
elif unavailable == "artifact":
|
|
monkeypatch.setattr(module, "_controller_release", lambda: None)
|
|
else:
|
|
monkeypatch.setattr(module, "REPOSITORY_ROOT", tmp_path / "missing")
|
|
status, payload, _ = request()
|
|
assert status == 503
|
|
assert payload["reason"] == "controller_runtime_prerequisite_unavailable"
|
|
for name in ("deploymentApply", "deploymentVerify", "rollback", "exactTargetPolicy", "enrollment", "targetRegistration"):
|
|
assert payload["features"][name] is False
|
|
|
|
|
|
def test_enrollment_recovers_when_safe_root_becomes_ready(controller, monkeypatch):
|
|
module, request = controller
|
|
assert request()[1]["features"]["enrollment"] is False
|
|
monkeypatch.setattr(module, "trusted_appdata_roots_readiness", lambda: [{"status": "ready"}])
|
|
status, payload, _ = request()
|
|
assert status == 200
|
|
assert payload["enrollmentReadiness"] == {"status": "ready", "reason": ""}
|
|
for name in ("enrollment", "targetRegistration", "runtimeSecretSealing"):
|
|
assert payload["features"][name] is True
|
|
|
|
|
|
@pytest.mark.parametrize("current,reason", [
|
|
({"status": "blocked", "checks": [{"name": "required_push_validation", "passed": False}]},
|
|
"controller_preflight_failed_required_push_validation"),
|
|
({"status": "preflight_ok", "currentRevision": "b" * 40, "checks": []},
|
|
"current_revision_changed"),
|
|
])
|
|
def test_self_upgrade_reports_failed_gate_before_missing_current_revision(controller, monkeypatch, current, reason):
|
|
module, _ = controller
|
|
monkeypatch.setattr(module, "_self_upgrade_preflight", lambda _revision: dict(current))
|
|
monkeypatch.setattr(module, "_self_upgrade_signature_is_valid", lambda *_args, **_kwargs: True)
|
|
intent = {"repository": module.SELF_UPGRADE_REPOSITORY, "revision": "c" * 40,
|
|
"reason": "Validated infrastructure repair", "expectedCurrentRevision": "a" * 40}
|
|
body = json.dumps(intent).encode()
|
|
handler = module.Handler.__new__(module.Handler)
|
|
handler.path = module.SELF_UPGRADE_PREFLIGHT_PATH
|
|
handler.headers = Message()
|
|
handler.headers["Content-Length"] = str(len(body))
|
|
handler.headers["Content-Type"] = "application/json"
|
|
handler.rfile = io.BytesIO(body)
|
|
handler._trusted_client = lambda: True
|
|
responses = []
|
|
handler._send_json = lambda status, payload: responses.append((status, json.loads(payload)))
|
|
handler._handle_self_upgrade()
|
|
assert responses[0][0] == 409
|
|
assert responses[0][1]["reason"] == reason
|