105 lines
2.9 KiB
Python
105 lines
2.9 KiB
Python
|
"""Integration tests: real ssh-keygen, no mocking.
|
||
|
|
|
||
|
|
Run with: uv run pytest -m integration
|
||
|
|
Excluded from the default unit suite by pyproject.toml addopts.
|
||
|
|
"""
|
||
|
|
import shutil
|
||
|
|
import subprocess
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from warden.ca import LocalCA, parse_cert_metadata
|
||
|
|
from warden.models import ActorType, CertSpec
|
||
|
|
|
||
|
|
pytestmark = pytest.mark.integration
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture(autouse=True)
|
||
|
|
def require_ssh_keygen():
|
||
|
|
if shutil.which("ssh-keygen") is None:
|
||
|
|
pytest.skip("ssh-keygen not found in PATH")
|
||
|
|
|
||
|
|
|
||
|
|
def _generate_ca_key(tmp_path: Path) -> Path:
|
||
|
|
ca_key = tmp_path / "ca_key"
|
||
|
|
subprocess.run(
|
||
|
|
["ssh-keygen", "-t", "ed25519", "-f", str(ca_key), "-N", "", "-C", "test-ca"],
|
||
|
|
check=True,
|
||
|
|
capture_output=True,
|
||
|
|
)
|
||
|
|
return ca_key
|
||
|
|
|
||
|
|
|
||
|
|
def _generate_actor_key(tmp_path: Path) -> Path:
|
||
|
|
key = tmp_path / "actor_key"
|
||
|
|
subprocess.run(
|
||
|
|
["ssh-keygen", "-t", "ed25519", "-f", str(key), "-N", "", "-C", "actor"],
|
||
|
|
check=True,
|
||
|
|
capture_output=True,
|
||
|
|
)
|
||
|
|
return key
|
||
|
|
|
||
|
|
|
||
|
|
def test_local_ca_sign_real_ssh_keygen(tmp_path):
|
||
|
|
ca_key = _generate_ca_key(tmp_path)
|
||
|
|
actor_key = _generate_actor_key(tmp_path)
|
||
|
|
pubkey = Path(str(actor_key) + ".pub")
|
||
|
|
|
||
|
|
spec = CertSpec(
|
||
|
|
actor_name="agt-integration-test",
|
||
|
|
actor_type=ActorType.AGT,
|
||
|
|
pubkey_path=pubkey,
|
||
|
|
ttl_hours=1,
|
||
|
|
principals=["agt-integration-test"],
|
||
|
|
identity="agt-integration-test",
|
||
|
|
)
|
||
|
|
|
||
|
|
state_dir = tmp_path / "state"
|
||
|
|
ca = LocalCA(ca_key, state_dir)
|
||
|
|
record = ca.sign(spec)
|
||
|
|
|
||
|
|
assert record.cert_path.exists()
|
||
|
|
assert record.valid_before > datetime.now(timezone.utc)
|
||
|
|
assert record.identity == "agt-integration-test"
|
||
|
|
assert record.principals == ["agt-integration-test"]
|
||
|
|
|
||
|
|
# Re-parse without any mocking
|
||
|
|
meta = parse_cert_metadata(record.cert_path)
|
||
|
|
assert meta["identity"] == "agt-integration-test"
|
||
|
|
assert meta["valid_before"] > datetime.now(timezone.utc)
|
||
|
|
|
||
|
|
|
||
|
|
def test_local_ca_sign_cert_file_mode_600(tmp_path):
|
||
|
|
ca_key = _generate_ca_key(tmp_path)
|
||
|
|
actor_key = _generate_actor_key(tmp_path)
|
||
|
|
pubkey = Path(str(actor_key) + ".pub")
|
||
|
|
|
||
|
|
spec = CertSpec(
|
||
|
|
actor_name="agt-integration-test",
|
||
|
|
actor_type=ActorType.AGT,
|
||
|
|
pubkey_path=pubkey,
|
||
|
|
ttl_hours=1,
|
||
|
|
principals=["agt-integration-test"],
|
||
|
|
identity="agt-integration-test",
|
||
|
|
)
|
||
|
|
|
||
|
|
ca = LocalCA(ca_key, tmp_path / "state")
|
||
|
|
record = ca.sign(spec)
|
||
|
|
|
||
|
|
assert oct(record.cert_path.stat().st_mode & 0o777) == oct(0o600)
|
||
|
|
|
||
|
|
|
||
|
|
def test_generate_keypair_real_ssh_keygen(tmp_path):
|
||
|
|
ca_key = _generate_ca_key(tmp_path)
|
||
|
|
ca = LocalCA(ca_key, tmp_path / "state")
|
||
|
|
|
||
|
|
privkey, pubkey = ca.generate_keypair("agt-integration-test")
|
||
|
|
|
||
|
|
assert privkey.exists()
|
||
|
|
assert pubkey.exists()
|
||
|
|
assert oct(privkey.stat().st_mode & 0o777) == oct(0o600)
|
||
|
|
assert oct(pubkey.stat().st_mode & 0o777) == oct(0o644)
|
||
|
|
assert "ssh-ed25519" in pubkey.read_text()
|