feat: add SSH key payload signing and verification endpoints
- POST /ssh-keys/{id}/sign - sign payload with Ed25519 private key
- POST /ssh-keys/{id}/verify - verify signature with public key
- Returns base64-encoded signatures
This commit is contained in:
@@ -1,3 +1,4 @@
|
||||
import base64
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
|
||||
@@ -74,6 +75,23 @@ class SSHKeyResponse(BaseModel):
|
||||
created_at: datetime
|
||||
|
||||
|
||||
class SignPayloadRequest(BaseModel):
|
||||
payload: str
|
||||
|
||||
|
||||
class SignatureResponse(BaseModel):
|
||||
signature: str
|
||||
|
||||
|
||||
class VerifySignatureRequest(BaseModel):
|
||||
payload: str
|
||||
signature: str
|
||||
|
||||
|
||||
class VerifySignatureResponse(BaseModel):
|
||||
valid: bool
|
||||
|
||||
|
||||
@router.post(
|
||||
"",
|
||||
response_model=SSHKeyResponse,
|
||||
@@ -166,3 +184,80 @@ async def delete_ssh_key(
|
||||
|
||||
await session.delete(ssh_key)
|
||||
await session.commit()
|
||||
|
||||
|
||||
@router.post(
|
||||
"/{key_id}/sign",
|
||||
response_model=SignatureResponse,
|
||||
summary="Sign payload",
|
||||
description="Sign a payload using the SSH private key.",
|
||||
)
|
||||
async def sign_payload(
|
||||
key_id: uuid.UUID,
|
||||
data: SignPayloadRequest,
|
||||
user_id: uuid.UUID = Depends(get_current_user_id),
|
||||
session: AsyncSession = Depends(get_db_session),
|
||||
) -> SignatureResponse:
|
||||
"""Sign a payload with an SSH key.
|
||||
|
||||
Args:
|
||||
key_id: UUID of the SSH key to use for signing.
|
||||
data: Sign request containing the payload string.
|
||||
user_id: ID of the authenticated user.
|
||||
session: Database session.
|
||||
|
||||
Returns:
|
||||
Base64-encoded Ed25519 signature.
|
||||
"""
|
||||
user = await _get_user(session, user_id)
|
||||
ssh_key = await session.get(SSHKey, key_id)
|
||||
if ssh_key is None or ssh_key.user_id != user.id:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="ssh key not found")
|
||||
|
||||
fernet = _get_fernet()
|
||||
private_key_pem = fernet.decrypt(ssh_key.private_key_encrypted.encode()).decode()
|
||||
|
||||
private_key = serialization.load_ssh_private_key(
|
||||
private_key_pem.encode(), password=None
|
||||
)
|
||||
|
||||
signature = private_key.sign(data.payload.encode())
|
||||
return SignatureResponse(signature=base64.b64encode(signature).decode())
|
||||
|
||||
|
||||
@router.post(
|
||||
"/{key_id}/verify",
|
||||
response_model=VerifySignatureResponse,
|
||||
summary="Verify signature",
|
||||
description="Verify a signature against a payload using the SSH public key.",
|
||||
)
|
||||
async def verify_signature(
|
||||
key_id: uuid.UUID,
|
||||
data: VerifySignatureRequest,
|
||||
user_id: uuid.UUID = Depends(get_current_user_id),
|
||||
session: AsyncSession = Depends(get_db_session),
|
||||
) -> VerifySignatureResponse:
|
||||
"""Verify a signature with an SSH key's public key.
|
||||
|
||||
Args:
|
||||
key_id: UUID of the SSH key to use for verification.
|
||||
data: Verify request containing payload and base64-encoded signature.
|
||||
user_id: ID of the authenticated user.
|
||||
session: Database session.
|
||||
|
||||
Returns:
|
||||
Whether the signature is valid.
|
||||
"""
|
||||
user = await _get_user(session, user_id)
|
||||
ssh_key = await session.get(SSHKey, key_id)
|
||||
if ssh_key is None or ssh_key.user_id != user.id:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="ssh key not found")
|
||||
|
||||
public_key = serialization.load_ssh_public_key(ssh_key.public_key.encode())
|
||||
|
||||
try:
|
||||
signature = base64.b64decode(data.signature)
|
||||
public_key.verify(signature, data.payload.encode())
|
||||
return VerifySignatureResponse(valid=True)
|
||||
except Exception:
|
||||
return VerifySignatureResponse(valid=False)
|
||||
|
||||
Reference in New Issue
Block a user