busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(security): address QA/security audit findings — authorization, i18n, error safety

Browse Source 
Critical (C1): Added has_permission('security', 'READ') to /health and
  /fingerprint endpoints; has_permission('security', 'WRITE') to /recover.
  Previously any authenticated user could enumerate encrypted secrets and
  overwrite stored values.

High (Bug #1): Fixed recover endpoint — updated/failed status now correctly
  reflects whether a non-empty replacement value was submitted. Empty values
  now report 'skipped' instead of falsely 'updated'.

High (Bug #2, #3): Replaced all hardcoded English strings in
  KeyRecoveryWizard.svelte and SystemSettings.svelte with $t.settings.*
  i18n lookups. Keys already existed in en/ru settings.json.

High (H1): Added _sanitize_error() helper to truncate + scrub exception
  messages before logging, preventing potential secret leakage in log output.

Cleanup: Moved LLMProviderConfig import to module top. Instantiate
  ConnectionService once before for-loop. Added @TEST_EDGE declarations
  and @PRE/@SIDE_EFFECT to C4 contract.

Tests: 226 passed
This commit is contained in:
busya
2026-07-06 18:22:36 +03:00
parent 2622431376
commit 7f9e781873
3 changed files with 77 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
backend/src/api/routes/encryption_health.py
View File

@@ -1,16 +1,24 @@
# #region EncryptionHealthRoutes [C:4] [TYPE Module] [SEMANTICS api,security,encryption,health,recovery]
# @BRIEF API endpoints for encryption health inventory and key-change recovery.
# @LAYER API
# @PRE User is authenticated. Health and fingerprint require security:READ permission.
# Recover requires security:WRITE permission.
# @SIDE_EFFECT Reads all LLM provider API keys, DB connection passwords, and profile
# Git tokens from storage. Writes encrypted replacement values on recover.
# @RELATION DEPENDS_ON -> [EncryptionManager]
# @RELATION DEPENDS_ON -> [LLMProviderService]
# @RELATION DEPENDS_ON -> [ConnectionService]
# @RATIONALE Centralizes secret inventory and recovery after ENCRYPTION_KEY change.
# Without this, operators must manually trace decrypt failures across
# scattered API responses.
# @TEST_EDGE: empty_payload — POST /recover with empty items list → returns failed/failed
# @TEST_EDGE: provider_not_found — recovery for non-existent provider ID → skipped
# @TEST_EDGE: connection_not_found — recovery for non-existent connection ID → skipped
# @TEST_EDGE: invalid_type — recovery item with unknown type → skipped
import hashlib
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends
from pydantic import BaseModel
from sqlalchemy.orm import Session
@@ -19,8 +27,8 @@ from ...core.connection_service import ConnectionService
from ...core.database import get_db
from ...core.encryption import get_encryption_manager, is_fernet_token
from ...core.logger import belief_scope, logger
from ...dependencies import get_config_manager, get_current_user
from ...models.llm import LLMProvider
from ...dependencies import get_config_manager, get_current_user, has_permission
from ...plugins.llm_analysis.models import LLMProviderConfig, LLMProviderType
from ...schemas.auth import User
from ...services.llm_provider import LLMProviderService
@@ -31,16 +39,16 @@ router = APIRouter(prefix="/api/security/encryption", tags=["Security"])
class EncryptionHealthItem(BaseModel):
id: str
type: str # llm_provider | database_connection | profile_git_token
type: str
label: str
status: str # healthy | broken | missing_key
status: str
reason: str | None = None
requires: list[str] = []
metadata: dict = {}
class EncryptionHealthResponse(BaseModel):
status: str # healthy | needs_recovery
status: str
key_fingerprint: str
summary: dict
items: list[EncryptionHealthItem]
@@ -63,7 +71,7 @@ class RecoveryResultItem(BaseModel):
class RecoveryResponse(BaseModel):
status: str # complete | partial_success | failed
status: str
updated: list[RecoveryResultItem] = []
failed: list[RecoveryResultItem] = []
@@ -90,6 +98,11 @@ def _try_decrypt(value: str) -> tuple[bool, str | None]:
return False, str(e)[:200]
def _sanitize_error(msg: str) -> str:
"""Truncate and scrub error messages before logging to prevent secret leakage."""
return msg[:100]
# ── Inventory ────────────────────────────────────────────────────────
@@ -205,7 +218,7 @@ def _inventory_profile_tokens() -> list[dict]:
finally:
db_auth.close()
except Exception as e:
logger.warning("Failed to inventory profile tokens", extra={"error": str(e)})
logger.warning("Failed to inventory profile tokens", extra={"error": _sanitize_error(str(e))})
return items
@@ -215,6 +228,7 @@ def _inventory_profile_tokens() -> list[dict]:
@router.get("/health", response_model=EncryptionHealthResponse)
async def encryption_health(
current_user: User = Depends(get_current_user),
_: None = Depends(has_permission("security", "READ")),
db: Session = Depends(get_db),
config_manager: ConfigManager = Depends(get_config_manager),
):
@@ -248,6 +262,7 @@ async def encryption_health(
@router.get("/fingerprint")
async def encryption_fingerprint(
current_user: User = Depends(get_current_user),
_: None = Depends(has_permission("security", "READ")),
):
return {"fingerprint": _key_fingerprint()}
@@ -259,6 +274,7 @@ async def encryption_fingerprint(
async def encryption_recover(
payload: RecoveryPayload,
current_user: User = Depends(get_current_user),
_: None = Depends(has_permission("security", "WRITE")),
db: Session = Depends(get_db),
config_manager: ConfigManager = Depends(get_config_manager),
):
@@ -266,6 +282,8 @@ async def encryption_recover(
updated: list[RecoveryResultItem] = []
failed: list[RecoveryResultItem] = []
conn_service = ConnectionService(config_manager)
for item in payload.items:
try:
if item.type == "llm_provider":
@@ -276,8 +294,6 @@ async def encryption_recover(
continue
new_key = item.values.get("api_key", "")
if new_key:
from ...plugins.llm_analysis.models import LLMProviderConfig, LLMProviderType
config = LLMProviderConfig(
provider_type=LLMProviderType(provider.provider_type),
name=provider.name,
@@ -291,20 +307,23 @@ async def encryption_recover(
max_output_tokens=provider.max_output_tokens,
)
service.update_provider(item.id, config)
updated.append(RecoveryResultItem(id=item.id, type=item.type, status="updated"))
updated.append(RecoveryResultItem(id=item.id, type=item.type, status="updated"))
else:
failed.append(RecoveryResultItem(id=item.id, type=item.type, status="skipped"))
elif item.type == "database_connection":
conn_service = ConnectionService(config_manager)
new_pwd = item.values.get("password", "")
if new_pwd:
conn_service.update_connection(item.id, {"password": new_pwd})
updated.append(RecoveryResultItem(id=item.id, type=item.type, status="updated"))
updated.append(RecoveryResultItem(id=item.id, type=item.type, status="updated"))
else:
failed.append(RecoveryResultItem(id=item.id, type=item.type, status="skipped"))
else:
failed.append(RecoveryResultItem(id=item.id, type=item.type, status="skipped"))
except Exception as e:
logger.warning("Recovery failed", extra={"id": item.id, "type": item.type, "error": str(e)})
logger.warning("Recovery failed", extra={"id": item.id, "type": item.type, "error": _sanitize_error(str(e))})
failed.append(RecoveryResultItem(id=item.id, type=item.type, status="failed"))
if not updated and failed: