alex
25db3f81b0
- Add DatabaseCredentialStorage with async CRUD operations - Create Credential SQLAlchemy model with encrypted values - Update GitCredential and AccessTokenCredential to support async - Fix SSH key encryption to use Fernet instead of base64 placeholder
83 lines
2.5 KiB
Python
83 lines
2.5 KiB
Python
"""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 uuid
|
|
from datetime import UTC, datetime
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.git.types import SshKeyStatus
|
|
|
|
|
|
def encrypt_private_key(raw: bytes) -> str:
|
|
from app.encryption import encrypt_value
|
|
|
|
return encrypt_value(raw.decode("utf-8"))
|
|
|
|
|
|
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(UTC))
|
|
updated_at: datetime = Field(default_factory=lambda: datetime.now(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(UTC)
|
|
if new_status == SshKeyStatus.revoked:
|
|
key.revoked_at = datetime.now(UTC)
|
|
return key
|