feat(FN-011): complete Step 3 — SSH key pair generation and lifecycle

Fusion-Task-Id: FN-011 Fusion-Task-Lineage: 4a9aca6f-9d91-43aa-8d2a-d59657c1541a
2026-05-14 07:36:35 +02:00
parent 2037359e8d
commit edfbce9635
2 changed files with 92 additions and 0 deletions
@@ -0,0 +1,89 @@
+"""SSH key pair generation and lifecycle management.
+
+Security rules:
+- Private key material must never appear in logs, exceptions, ``__repr__``,
+  or test output.
+- The ``encrypted_private_key`` field uses ``repr=False``.
+"""
+
+import base64
+import uuid
+from datetime import datetime, timezone
+
+from pydantic import BaseModel, Field
+
+from app.git.types import SshKeyStatus
+
+
+def encrypt_private_key(raw: bytes) -> str:
+    """Placeholder encryption helper.
+
+    base64-encodes *raw* until FN-009 delivers the real encryption backend.
+    """
+    return base64.b64encode(raw).decode("ascii")
+
+
+class SshKeyPair(BaseModel):
+    """An Ed25519 SSH key pair belonging to a repository connection."""
+
+    id: uuid.UUID = Field(default_factory=uuid.uuid4)
+    connection_id: uuid.UUID
+    public_key: str
+    encrypted_private_key: str = Field(repr=False)
+    status: SshKeyStatus = SshKeyStatus.generated
+    created_at: datetime = Field(
+        default_factory=lambda: datetime.now(timezone.utc)
+    )
+    updated_at: datetime = Field(
+        default_factory=lambda: datetime.now(timezone.utc)
+    )
+    revoked_at: datetime | None = None
+
+
+class SshKeyLifecycle:
+    """Generate and transition SSH key pairs."""
+
+    @staticmethod
+    def generate(connection_id: uuid.UUID) -> SshKeyPair:
+        """Generate a new Ed25519 key pair for *connection_id*."""
+        from cryptography.hazmat.primitives.asymmetric.ed25519 import (
+            Ed25519PrivateKey,
+        )
+        from cryptography.hazmat.primitives.serialization import (
+            Encoding,
+            NoEncryption,
+            PrivateFormat,
+            PublicFormat,
+        )
+
+        private_key = Ed25519PrivateKey.generate()
+        public_key = private_key.public_key()
+
+        public_key_pem = public_key.public_bytes(
+            Encoding.OpenSSH, PublicFormat.OpenSSH
+        ).decode("utf-8")
+
+        private_key_pem = private_key.private_bytes(
+            Encoding.PEM, PrivateFormat.PKCS8, NoEncryption()
+        )
+
+        encrypted = encrypt_private_key(private_key_pem)
+
+        return SshKeyPair(
+            connection_id=connection_id,
+            public_key=public_key_pem,
+            encrypted_private_key=encrypted,
+            status=SshKeyStatus.generated,
+        )
+
+    @staticmethod
+    def transition(key: SshKeyPair, new_status: SshKeyStatus) -> SshKeyPair:
+        """Update *key* status and timestamps.
+
+        Sets ``revoked_at`` when transitioning to :attr:`SshKeyStatus.revoked`.
+        """
+        key.status = new_status
+        key.updated_at = datetime.now(timezone.utc)
+        if new_status == SshKeyStatus.revoked:
+            key.revoked_at = datetime.now(timezone.utc)
+        return key