vulncheck
cab46fd819
Stage 4 of the score cascade (vulnrichment → NVD → cvelistV5 → GHSA). GHSA carries advisories for CVEs still missing from NVD and cvelistV5 when very fresh — the gap the tester hit on new Firefox/Notepad++ CVEs. Fills CVSS/severity/description/references only; GHSA 'unreviewed' advisories carry no affected-version range (verified against the live API), so no fix/detection data is derived — this is enrichment, not new-CVE detection. Optional github_pat setting (encrypted at rest, admin-only Settings card) lifts the GitHub rate limit 60 → 5000 req/h; the loop stops cleanly when the limit is hit.
86 lines
2.5 KiB
Python
86 lines
2.5 KiB
Python
"""
|
|
Transparent at-rest encryption for sensitive setting values.
|
|
|
|
Setting rows whose `key` is in PROTECTED_SETTING_KEYS get their `value`
|
|
column encrypted with the AUTH_PROVIDER_CRYPTO_KEY Fernet key (same key
|
|
that protects TOTP secrets). Encrypted values are prefixed with
|
|
`ENC_PREFIX` so reads can detect ciphertext vs. legacy plaintext and
|
|
auto-migrate on the next write.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Optional
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.auth.totp import _fernet # reuses AUTH_PROVIDER_CRYPTO_KEY
|
|
from app.models.setting import Setting
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
ENC_PREFIX = "enc:v1:"
|
|
|
|
# Settings whose value field contains secrets (API tokens, passwords).
|
|
# Values written under these keys are encrypted at rest; reads transparently
|
|
# decrypt. Pre-existing plaintext rows still readable until next write.
|
|
PROTECTED_SETTING_KEYS: frozenset[str] = frozenset({
|
|
"wazuh_config",
|
|
"smtp_config",
|
|
"nessus_config",
|
|
"openrouter_api_key",
|
|
"intune_config",
|
|
"github_pat",
|
|
})
|
|
|
|
|
|
def is_protected(key: str) -> bool:
|
|
return key in PROTECTED_SETTING_KEYS
|
|
|
|
|
|
def encrypt_value(plaintext: str) -> str:
|
|
"""Return ENC_PREFIX + base64-ciphertext."""
|
|
token = _fernet().encrypt(plaintext.encode()).decode()
|
|
return f"{ENC_PREFIX}{token}"
|
|
|
|
|
|
def decrypt_value(stored: str) -> str:
|
|
"""Decrypt an ENC_PREFIX-tagged value. Passes plaintext through unchanged."""
|
|
if not stored or not stored.startswith(ENC_PREFIX):
|
|
return stored
|
|
ciphertext = stored[len(ENC_PREFIX):]
|
|
return _fernet().decrypt(ciphertext.encode()).decode()
|
|
|
|
|
|
def read_setting_value(db: Session, key: str) -> Optional[str]:
|
|
"""Return decrypted setting value (or None if not set)."""
|
|
row = db.query(Setting).filter(Setting.key == key).first()
|
|
if not row or row.value is None:
|
|
return None
|
|
if is_protected(key):
|
|
try:
|
|
return decrypt_value(row.value)
|
|
except Exception as e:
|
|
logger.error("Failed to decrypt setting %s: %s", key, e)
|
|
return None
|
|
return row.value
|
|
|
|
|
|
def write_setting_value(
|
|
db: Session,
|
|
key: str,
|
|
value: str,
|
|
description: str = "",
|
|
) -> Setting:
|
|
"""Upsert a setting, encrypting if the key is protected."""
|
|
stored = encrypt_value(value) if is_protected(key) else value
|
|
row = db.query(Setting).filter(Setting.key == key).first()
|
|
if row:
|
|
row.value = stored
|
|
else:
|
|
row = Setting(key=key, value=stored, description=description)
|
|
db.add(row)
|
|
db.commit()
|
|
db.refresh(row)
|
|
return row
|