coryHawkvelt/gatehouse-api
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(ssh): change SSH key uniqueness to per-user scope

Browse Source 
Previously, SSH key fingerprints were globally unique across all users,
preventing the same key from being registered by different users. This
change makes fingerprint uniqueness scoped to individual users.

- Remove global unique constraints on payload and fingerprint columns
- Add composite unique constraint on (user_id, fingerprint)
- Make add_ssh_key operation idempotent for same user
- Return tuple (SSHKey, is_new) from service to indicate creation status
- Update API to return 200 for existing keys, 201 for new keys

BREAKING CHANGE: API behavior changed - duplicate key addition now
returns 200 OK instead of 409 Conflict. Service method signature changed
from returning SSHKey to tuple[SSHKey, bool].
This commit is contained in:
Cory Hawkvelt
2026-04-25 06:22:08 +09:30
parent cec04f3cb2
commit de6f39e7e3
5 changed files with 132 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gatehouse_app/api/v1/ssh/keys.py
+6 -5
View File
@@ -32,11 +32,12 @@ def add_ssh_key():
return api_response(success=False, message='public_key is required', status=400, error_type='BAD_REQUEST')
try:
ssh_key = ssh_key_service.add_ssh_key(user_id=user_id, public_key=public_key, description=description)
AuditLog.log(action=AuditAction.SSH_KEY_ADDED, user_id=user_id, resource_type='SSHKey', resource_id=ssh_key.id, ip_address=request.remote_addr)
return api_response(success=True, message='SSH key added', data=ssh_key.to_dict(), status=201)
except SSHKeyAlreadyExistsError as e:
return api_response(success=False, message=e.message, status=409, error_type='SSH_KEY_ALREADY_EXISTS')
ssh_key, is_new = ssh_key_service.add_ssh_key(user_id=user_id, public_key=public_key, description=description)
if is_new:
AuditLog.log(action=AuditAction.SSH_KEY_ADDED, user_id=user_id, resource_type='SSHKey', resource_id=ssh_key.id, ip_address=request.remote_addr)
return api_response(success=True, message='SSH key added', data=ssh_key.to_dict(), status=201)
else:
return api_response(success=True, message='SSH key already exists', data=ssh_key.to_dict(), status=200)
except IntegrityError:
return api_response(success=False, message='SSH key already exists', status=409, error_type='SSH_KEY_ALREADY_EXISTS')
except SSHKeyError as e:
gatehouse_app/models/ssh_ca/ssh_key.py
+3 -2
View File
@@ -21,10 +21,10 @@ class SSHKey(BaseModel):
)
# SSH key payload in OpenSSH format (e.g., "ssh-ed25519 AAAAB3Nz...")
payload = db.Column(db.Text, nullable=False, unique=True)
payload = db.Column(db.Text, nullable=False)
# SHA256 fingerprint for quick comparison and deduplication
fingerprint = db.Column(db.String(255), nullable=False, unique=True, index=True)
fingerprint = db.Column(db.String(255), nullable=False, index=True)
# Optional human-readable description (e.g., "My laptop key")
description = db.Column(db.String(255), nullable=True)
@@ -53,6 +53,7 @@ class SSHKey(BaseModel):
__table_args__ = (
db.Index("idx_ssh_key_user_verified", "user_id", "verified"),
db.UniqueConstraint('user_id', 'fingerprint', name='uix_user_fingerprint'),
)
def __repr__(self):
gatehouse_app/services/ssh_key_service.py
+26 -22
View File
@@ -41,46 +41,47 @@ class SSHKeyService:
user_id: str,
public_key: str,
description: Optional[str] = None,
) -> SSHKey:
) -> tuple[SSHKey, bool]:
"""Add an SSH public key for a user.
Args:
user_id: ID of the user
public_key: SSH public key in OpenSSH format
description: Optional description of the key
Returns:
Created SSHKey instance
Tuple of (SSHKey instance, is_new) where is_new is True for
newly created keys, False for existing keys (idempotent).
Raises:
UserNotFoundError: If user doesn't exist
SSHKeyError: If key format is invalid
SSHKeyAlreadyExistsError: If key already exists
"""
# Verify user exists
user = User.query.get(user_id)
if not user:
raise UserNotFoundError(f"User {user_id} not found")
# Validate key format
if not verify_ssh_key_format(public_key):
raise SSHKeyError("Invalid SSH public key format")
# Compute fingerprint
try:
fingerprint = compute_ssh_fingerprint(public_key)
except Exception as e:
logger.error(f"Failed to compute fingerprint: {str(e)}")
raise SSHKeyError(f"Failed to compute key fingerprint: {str(e)}")
# Check for duplicate (including soft-deleted records — fingerprint is unique in DB)
existing = SSHKey.query.filter_by(fingerprint=fingerprint).first()
# Check for duplicate per user (including soft-deleted records)
existing = SSHKey.query.filter_by(
user_id=user_id, fingerprint=fingerprint
).first()
if existing:
if existing.deleted_at is not None:
# Restore the soft-deleted key: clear deleted_at and update fields
existing.deleted_at = None
existing.user_id = user_id
existing.description = description or existing.description
existing.description = description if description is not None else existing.description
existing.verified = False
existing.verified_at = None
existing.verify_text = None
@@ -90,15 +91,18 @@ class SSHKeyService:
f"Restored soft-deleted SSH key for user {user_id}: "
f"fingerprint={fingerprint}"
)
return existing
raise SSHKeyAlreadyExistsError(
f"SSH key with fingerprint {fingerprint} already exists"
return existing, False
# Idempotent: return existing key without error
logger.info(
f"SSH key already exists for user {user_id}: "
f"fingerprint={fingerprint}"
)
return existing, False
# Extract metadata
key_type = extract_ssh_key_type(public_key)
key_comment = extract_ssh_key_comment(public_key)
# Create SSH key record
ssh_key = SSHKey(
user_id=user_id,
@@ -109,15 +113,15 @@ class SSHKeyService:
key_comment=key_comment,
verified=False,
)
ssh_key.save()
logger.info(
f"SSH key added for user {user_id}: "
f"fingerprint={fingerprint}, type={key_type}"
)
return ssh_key
return ssh_key, True
def get_ssh_key(self, key_id: str) -> SSHKey:
"""Get an SSH key by ID.