stroblmeandClaude Opus 5.5
38013a0774
Docs / docs (push) Successful in 32s
Playwright Tests / test-playwright (1, 2) (push) Failing after 6m58s
Playwright Tests / test-playwright (2, 2) (push) Failing after 2m33s
pre-commit / pre-commit (push) Failing after 3m1s
Test Backend / test-backend (push) Failing after 3m13s
Compose Smoke Test / test-compose (push) Successful in 31s
Playwright Tests / merge-reports (push) Successful in 1m34s
A pin trusts one exact certificate, so an expired one or an engine clock that is off no longer locks a device out (X509_V_FLAG_NO_CHECK_TIME). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013GAeg4Jo6xjZzjwy9QDWMQ
111 lines
3.4 KiB
Python
111 lines
3.4 KiB
Python
"""A pinned certificate is trusted for what it is, not for its dates."""
|
|
|
|
import datetime
|
|
import socket
|
|
import ssl
|
|
import threading
|
|
from collections.abc import Iterator
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from cryptography import x509
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
|
from cryptography.hazmat.primitives.asymmetric import ec
|
|
from cryptography.x509.oid import NameOID
|
|
|
|
from fluksio.flow.connector_tls import (
|
|
CertificateChanged,
|
|
context_for,
|
|
pin_key,
|
|
pinned_certificate,
|
|
raise_if_changed,
|
|
)
|
|
|
|
|
|
def _expired(name: str) -> tuple[ec.EllipticCurvePrivateKey, str]:
|
|
"""A self-signed certificate that expired a month ago."""
|
|
key = ec.generate_private_key(ec.SECP256R1())
|
|
now = datetime.datetime.now(datetime.UTC)
|
|
subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, name)])
|
|
certificate = (
|
|
x509.CertificateBuilder()
|
|
.subject_name(subject)
|
|
.issuer_name(subject)
|
|
.public_key(key.public_key())
|
|
.serial_number(x509.random_serial_number())
|
|
.not_valid_before(now - datetime.timedelta(days=60))
|
|
.not_valid_after(now - datetime.timedelta(days=30))
|
|
.sign(key, hashes.SHA256())
|
|
)
|
|
return key, certificate.public_bytes(serialization.Encoding.PEM).decode()
|
|
|
|
|
|
DEVICE = _expired("00005e005301")
|
|
IMPOSTOR = _expired("00005e005302")
|
|
|
|
|
|
class Store:
|
|
def __init__(self) -> None:
|
|
self.values: dict[str, str] = {}
|
|
|
|
def get(self, key: str) -> str | None:
|
|
return self.values.get(key)
|
|
|
|
def set(self, key: str, value: str) -> None:
|
|
self.values[key] = value
|
|
|
|
|
|
@pytest.fixture
|
|
def device(tmp_path: Path) -> Iterator[str]:
|
|
"""A TLS server presenting the expired certificate; yields its address."""
|
|
key, pem = DEVICE
|
|
(tmp_path / "cert.pem").write_text(pem)
|
|
(tmp_path / "key.pem").write_bytes(
|
|
key.private_bytes(
|
|
serialization.Encoding.PEM,
|
|
serialization.PrivateFormat.PKCS8,
|
|
serialization.NoEncryption(),
|
|
)
|
|
)
|
|
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
|
|
context.load_cert_chain(tmp_path / "cert.pem", tmp_path / "key.pem")
|
|
listener = socket.create_server(("127.0.0.1", 0))
|
|
|
|
def serve() -> None:
|
|
while True:
|
|
try:
|
|
connection, _ = listener.accept()
|
|
except OSError:
|
|
return
|
|
try:
|
|
context.wrap_socket(connection, server_side=True).close()
|
|
except OSError:
|
|
connection.close()
|
|
|
|
threading.Thread(target=serve, daemon=True).start()
|
|
yield f"127.0.0.1:{listener.getsockname()[1]}"
|
|
listener.close()
|
|
|
|
|
|
def _handshake(address: str, pem: str) -> None:
|
|
host, port = address.split(":")
|
|
with socket.create_connection((host, int(port)), timeout=5) as raw:
|
|
context_for(pem).wrap_socket(raw, server_hostname=host).close()
|
|
|
|
|
|
def test_an_expired_pinned_certificate_is_still_trusted(device: str) -> None:
|
|
store = Store()
|
|
pem = pinned_certificate(store, device, 5.0)
|
|
|
|
assert store.get(pin_key(device)) == DEVICE[1]
|
|
_handshake(device, pem) # raises if the dates were checked
|
|
|
|
|
|
def test_another_certificate_is_still_refused(device: str) -> None:
|
|
pinned = IMPOSTOR[1]
|
|
with pytest.raises(ssl.SSLCertVerificationError) as refused:
|
|
_handshake(device, pinned)
|
|
|
|
with pytest.raises(CertificateChanged, match="/connectors/test/state/cert:"):
|
|
raise_if_changed(refused.value, pinned, device, 5.0, "test")
|