Clean Core security audit findings
This commit is contained in:
@@ -201,7 +201,7 @@ class AuthGroupsResponse(BaseModel):
|
||||
|
||||
class LoginResponse(BaseModel):
|
||||
access_token: str
|
||||
token_type: str = "bearer"
|
||||
token_type: str = "bearer" # noqa: S105 - OAuth token type, not a credential.
|
||||
expires_at: datetime
|
||||
user: UserInfo
|
||||
# Backwards-compatible alias for the active tenant.
|
||||
|
||||
@@ -1,5 +1,3 @@
|
||||
from __future__ import annotations
|
||||
|
||||
"""Core auth dependency facade.
|
||||
|
||||
Routers depend on this module instead of a concrete access-provider package.
|
||||
@@ -7,6 +5,8 @@ The active auth module provides the request principal through the platform
|
||||
capability registry.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
from fastapi import Depends, Header, HTTPException, Request, status
|
||||
|
||||
@@ -1,8 +1,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from collections.abc import Mapping
|
||||
from typing import Any, Literal
|
||||
from typing import Literal
|
||||
|
||||
from govoplan_core.security.permissions import scopes_grant
|
||||
from govoplan_core.security.redaction import contains_plain_secret
|
||||
@@ -22,7 +21,7 @@ class ConfigurationFieldSafety:
|
||||
storage: str
|
||||
ui_managed: bool
|
||||
risk: ConfigurationRisk
|
||||
secret_handling: SecretHandling = "none"
|
||||
secret_handling: SecretHandling = "none" # noqa: S105 - policy vocabulary.
|
||||
required_scopes: tuple[str, ...] = ()
|
||||
dry_run_required: bool = False
|
||||
validation_required: bool = True
|
||||
@@ -69,7 +68,7 @@ class ConfigurationChangeSafetyPlan:
|
||||
maintenance_required: bool = False
|
||||
maintenance_satisfied: bool = False
|
||||
rollback_history_required: bool = False
|
||||
secret_handling: SecretHandling = "none"
|
||||
secret_handling: SecretHandling = "none" # noqa: S105 - policy vocabulary.
|
||||
audit_event: str | None = None
|
||||
policy_explanation: str | None = None
|
||||
blockers: tuple[str, ...] = ()
|
||||
@@ -240,7 +239,7 @@ _CONFIGURATION_FIELD_SAFETY: tuple[ConfigurationFieldSafety, ...] = (
|
||||
storage="module_settings",
|
||||
ui_managed=True,
|
||||
risk="high",
|
||||
secret_handling="reference_only",
|
||||
secret_handling="reference_only", # noqa: S106 # nosec B106 - policy vocabulary.
|
||||
required_scopes=("mail_servers:manage_credentials",),
|
||||
validation_required=True,
|
||||
audit_event="mail_server_profile.credential_updated",
|
||||
@@ -256,7 +255,7 @@ _CONFIGURATION_FIELD_SAFETY: tuple[ConfigurationFieldSafety, ...] = (
|
||||
storage="module_settings",
|
||||
ui_managed=True,
|
||||
risk="high",
|
||||
secret_handling="reference_only",
|
||||
secret_handling="reference_only", # noqa: S106 # nosec B106 - policy vocabulary.
|
||||
required_scopes=("files:file:admin",),
|
||||
dry_run_required=True,
|
||||
policy_explanation_required=True,
|
||||
@@ -273,7 +272,7 @@ _CONFIGURATION_FIELD_SAFETY: tuple[ConfigurationFieldSafety, ...] = (
|
||||
storage="environment",
|
||||
ui_managed=False,
|
||||
risk="destructive",
|
||||
secret_handling="env_only",
|
||||
secret_handling="env_only", # noqa: S106 # nosec B106 - policy vocabulary.
|
||||
maintenance_required=True,
|
||||
notes="Database connectivity remains deployment-managed and must not be changed from the running UI.",
|
||||
),
|
||||
@@ -285,7 +284,7 @@ _CONFIGURATION_FIELD_SAFETY: tuple[ConfigurationFieldSafety, ...] = (
|
||||
storage="environment",
|
||||
ui_managed=False,
|
||||
risk="destructive",
|
||||
secret_handling="env_only",
|
||||
secret_handling="env_only", # noqa: S106 # nosec B106 - policy vocabulary.
|
||||
maintenance_required=True,
|
||||
two_person_approval_required=True,
|
||||
notes="Encryption roots remain out of band; UI may only report missing/rotated state.",
|
||||
@@ -298,7 +297,7 @@ _CONFIGURATION_FIELD_SAFETY: tuple[ConfigurationFieldSafety, ...] = (
|
||||
storage="environment",
|
||||
ui_managed=False,
|
||||
risk="high",
|
||||
secret_handling="env_only",
|
||||
secret_handling="env_only", # noqa: S106 # nosec B106 - policy vocabulary.
|
||||
notes="Trust roots are deployment-managed; UI can validate catalogs but should not edit key material.",
|
||||
),
|
||||
ConfigurationFieldSafety(
|
||||
@@ -309,7 +308,7 @@ _CONFIGURATION_FIELD_SAFETY: tuple[ConfigurationFieldSafety, ...] = (
|
||||
storage="environment",
|
||||
ui_managed=False,
|
||||
risk="high",
|
||||
secret_handling="env_only",
|
||||
secret_handling="env_only", # noqa: S106 # nosec B106 - policy vocabulary.
|
||||
notes="Configuration package trust roots are deployment-managed.",
|
||||
),
|
||||
)
|
||||
@@ -414,9 +413,9 @@ def _configuration_change_safety_state(
|
||||
approval_satisfied = not approval_required or approval_count >= 2
|
||||
if approval_required and not approval_satisfied:
|
||||
blockers.append("two_person_approval_required")
|
||||
if field.secret_handling == "reference_only" and _contains_plain_secret(value):
|
||||
if field.secret_handling == "reference_only" and _contains_plain_secret(value): # noqa: S105 # nosec B105 - policy vocabulary.
|
||||
blockers.append("secret_reference_required")
|
||||
if field.secret_handling == "env_only" and value is not None:
|
||||
if field.secret_handling == "env_only" and value is not None: # noqa: S105 # nosec B105 - policy vocabulary.
|
||||
blockers.append("env_only_secret")
|
||||
if field.rollback_history_required:
|
||||
warnings.append("rollback_history_required")
|
||||
@@ -466,7 +465,7 @@ def _policy_explanation(field: ConfigurationFieldSafety) -> str:
|
||||
parts.append("requires two-person approval")
|
||||
if field.maintenance_required:
|
||||
parts.append("requires maintenance mode")
|
||||
if field.secret_handling != "none":
|
||||
if field.secret_handling != "none": # noqa: S105 # nosec B105 - policy vocabulary.
|
||||
parts.append(f"uses {field.secret_handling} secret handling")
|
||||
return "; ".join(parts) + "."
|
||||
|
||||
|
||||
@@ -106,7 +106,8 @@ def installer_notification_body(event_kind: str, request: Mapping[str, object])
|
||||
run_id = _result_run_id(request)
|
||||
if run_id:
|
||||
return ". Run: ".join((sentence, run_id))
|
||||
return sentence + "."
|
||||
# This helper returns plain notification text, not an HTTP/HTML response.
|
||||
return sentence + "." # nosemgrep: python.flask.security.audit.directly-returned-format-string.directly-returned-format-string
|
||||
|
||||
|
||||
def installer_notification_priority(status: str) -> int:
|
||||
|
||||
@@ -6,7 +6,6 @@ from datetime import UTC, datetime
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from cryptography.exceptions import InvalidSignature
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
|
||||
@@ -65,7 +65,7 @@ def bootstrap_dev_data(
|
||||
api_key_secret: str | None = None,
|
||||
tenant_slug: str = "default",
|
||||
user_email: str = "admin@example.local",
|
||||
user_password: str = "dev-admin",
|
||||
user_password: str = "dev-admin", # noqa: S107 - development bootstrap only.
|
||||
) -> BootstrapResult:
|
||||
tenant = session.query(Tenant).filter(Tenant.slug == tenant_slug).one_or_none()
|
||||
if tenant is None:
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
from collections.abc import Iterable, Mapping
|
||||
from dataclasses import dataclass, replace
|
||||
import json
|
||||
import logging
|
||||
@@ -634,7 +634,7 @@ def _backfill_user_lock_state_for_create_all_schema(database_url: str) -> None:
|
||||
|
||||
def _row_count(connection, table_name: str) -> int:
|
||||
quoted = _quoted_table_name(connection, table_name)
|
||||
statement = text(f"SELECT COUNT(*) FROM {quoted}") # nosec B608 # nosemgrep: python.sqlalchemy.security.audit.avoid-sqlalchemy-text.avoid-sqlalchemy-text
|
||||
statement = text(f"SELECT COUNT(*) FROM {quoted}") # noqa: S608 # nosec B608 # nosemgrep: python.sqlalchemy.security.audit.avoid-sqlalchemy-text.avoid-sqlalchemy-text
|
||||
return int(connection.execute(statement).scalar_one())
|
||||
|
||||
|
||||
|
||||
@@ -41,7 +41,7 @@ def fetch_http(
|
||||
method: str = "GET",
|
||||
headers: Mapping[str, str] | None = None,
|
||||
) -> HttpFetchResponse:
|
||||
request = urllib.request.Request(
|
||||
request = urllib.request.Request( # noqa: S310 - URL is restricted to validated HTTP(S).
|
||||
validate_http_url(url, label=label),
|
||||
headers=dict(headers or {}),
|
||||
method=method,
|
||||
|
||||
@@ -2,7 +2,7 @@ from __future__ import annotations
|
||||
|
||||
import re
|
||||
from collections.abc import Mapping
|
||||
from typing import Any
|
||||
|
||||
|
||||
def sensitive_key_tokens(key: object) -> set[str]:
|
||||
value = str(key).strip()
|
||||
@@ -40,7 +40,7 @@ def contains_plain_secret(value: object) -> bool:
|
||||
normalized_key = str(key).strip().casefold().replace("-", "_")
|
||||
if normalized_key in {"credential_ref", "secret_ref", "secret_reference"}:
|
||||
continue
|
||||
if is_sensitive_key(key) and item not in (None, "", {"secret_ref": ""}):
|
||||
if is_sensitive_key(key) and item not in (None, "", {"secret_ref": ""}): # nosec B105 - empty redaction sentinels.
|
||||
return True
|
||||
if isinstance(item, Mapping) and contains_plain_secret(item):
|
||||
return True
|
||||
|
||||
@@ -18,7 +18,7 @@ class SecretDecryptionError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
CAPABILITY_SECURITY_SECRET_PROVIDER = "security.secretProvider"
|
||||
CAPABILITY_SECURITY_SECRET_PROVIDER = "security.secretProvider" # noqa: S105 # nosec B105 - capability identifier.
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
|
||||
Reference in New Issue
Block a user