200 lines
7.6 KiB
Python
200 lines
7.6 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
from collections.abc import Mapping
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
_SECRET_ENV_ALLOWLIST = "GOVOPLAN_CONNECTOR_SECRET_ENV_ALLOWLIST" # noqa: S105 # nosec B105 - configuration key.
|
|
_CA_BUNDLE_ALLOWLIST = "GOVOPLAN_CONNECTOR_CA_BUNDLE_ALLOWLIST"
|
|
_ENV_NAME = re.compile(r"[A-Za-z_][A-Za-z0-9_]*\Z")
|
|
_SECRET_ENV_METADATA_KEYS = frozenset(
|
|
{
|
|
"access_key_id_env",
|
|
"secret_access_key_env",
|
|
"session_token_env",
|
|
}
|
|
)
|
|
_DEVELOPMENT_ENVIRONMENTS = frozenset({"dev", "development", "local", "test", "testing"})
|
|
|
|
|
|
class ConnectorDeploymentConfigurationError(ValueError):
|
|
"""Raised when connector data crosses a deployment-owned trust boundary."""
|
|
|
|
|
|
def connector_secret_env_value(name: str, *, source_kind: str) -> str:
|
|
clean_name = _validate_secret_env_reference(name, source_kind=source_kind)
|
|
value = os.environ.get(clean_name)
|
|
if value is None or value == "":
|
|
raise ConnectorDeploymentConfigurationError("Connector credential environment variable is not configured")
|
|
return value
|
|
|
|
|
|
def connector_secret_env_available(name: str | None, *, source_kind: str) -> bool:
|
|
if not name:
|
|
return False
|
|
try:
|
|
clean_name = _validate_secret_env_reference(name, source_kind=source_kind)
|
|
except ConnectorDeploymentConfigurationError:
|
|
return False
|
|
return bool(os.environ.get(clean_name))
|
|
|
|
|
|
def validate_deployment_connector_references(
|
|
*,
|
|
source_kind: str,
|
|
password_env: str | None = None,
|
|
token_env: str | None = None,
|
|
metadata: Mapping[str, Any] | None = None,
|
|
) -> None:
|
|
"""Validate references in deployment-owned connector configuration."""
|
|
|
|
for name in (password_env, token_env):
|
|
if name:
|
|
_validate_secret_env_reference(name, source_kind=source_kind)
|
|
for key in _SECRET_ENV_METADATA_KEYS:
|
|
name = _clean((metadata or {}).get(key))
|
|
if name:
|
|
_validate_secret_env_reference(name, source_kind=source_kind)
|
|
validate_connector_tls_metadata(metadata)
|
|
|
|
|
|
def reject_api_controlled_deployment_references(
|
|
*,
|
|
password_env: str | None = None,
|
|
token_env: str | None = None,
|
|
metadata: Mapping[str, Any] | None = None,
|
|
) -> None:
|
|
"""Reject process-secret selectors controlled through tenant-facing APIs."""
|
|
|
|
if _clean(password_env) or _clean(token_env):
|
|
raise ConnectorDeploymentConfigurationError(
|
|
"Environment-backed credentials may only be declared in deployment-owned connector configuration"
|
|
)
|
|
for key, value in (metadata or {}).items():
|
|
if _clean(value) and (str(key).strip().casefold() in _SECRET_ENV_METADATA_KEYS or str(key).strip().casefold().endswith("_env")):
|
|
raise ConnectorDeploymentConfigurationError(
|
|
"Environment-backed credentials may only be declared in deployment-owned connector configuration"
|
|
)
|
|
validate_connector_tls_metadata(metadata)
|
|
|
|
|
|
def connector_ca_bundle_path(value: str) -> str:
|
|
clean_value = _clean(value)
|
|
if not clean_value:
|
|
raise ConnectorDeploymentConfigurationError("Connector CA bundle path is empty")
|
|
candidate = Path(clean_value)
|
|
if not candidate.is_absolute():
|
|
raise ConnectorDeploymentConfigurationError("Connector CA bundle paths must be absolute")
|
|
try:
|
|
resolved = candidate.resolve(strict=True)
|
|
except OSError as exc:
|
|
raise ConnectorDeploymentConfigurationError("Connector CA bundle path does not exist") from exc
|
|
allowed = _allowed_ca_bundle_paths()
|
|
if resolved not in allowed:
|
|
raise ConnectorDeploymentConfigurationError(
|
|
f"Connector CA bundle path is not listed in {_CA_BUNDLE_ALLOWLIST}"
|
|
)
|
|
if not resolved.is_file():
|
|
raise ConnectorDeploymentConfigurationError("Connector CA bundle path must be a regular file")
|
|
return str(resolved)
|
|
|
|
|
|
def validate_connector_tls_metadata(metadata: Mapping[str, Any] | None) -> None:
|
|
values = metadata or {}
|
|
ca_bundle = _clean(values.get("ca_bundle"))
|
|
if ca_bundle:
|
|
connector_ca_bundle_path(ca_bundle)
|
|
for key in ("verify_tls", "tls_verify"):
|
|
if key in values and not _as_bool(values.get(key), default=True) and not _development_runtime():
|
|
raise ConnectorDeploymentConfigurationError(
|
|
"Connector TLS certificate verification may only be disabled in dev/test environments"
|
|
)
|
|
|
|
|
|
def connector_effective_endpoint_url(
|
|
*,
|
|
provider: str | None,
|
|
endpoint_url: str | None,
|
|
metadata: Mapping[str, Any] | None,
|
|
) -> str | None:
|
|
"""Return the endpoint that connector I/O will actually use."""
|
|
|
|
clean_provider = (provider or "").strip().casefold()
|
|
values = metadata or {}
|
|
webdav_url = _clean(values.get("webdav_endpoint_url"))
|
|
browse_protocol = (_clean(values.get("browse_protocol")) or "").casefold()
|
|
if webdav_url and (clean_provider in {"seafile", "webdav", "nextcloud"} or browse_protocol == "webdav"):
|
|
return webdav_url
|
|
return _clean(endpoint_url)
|
|
|
|
|
|
def _validate_secret_env_reference(name: str, *, source_kind: str) -> str:
|
|
if source_kind.strip().casefold() != "settings":
|
|
raise ConnectorDeploymentConfigurationError(
|
|
"Environment-backed credentials may only be used by deployment-owned connector configuration"
|
|
)
|
|
clean_name = name.strip()
|
|
if not _ENV_NAME.fullmatch(clean_name):
|
|
raise ConnectorDeploymentConfigurationError("Connector credential environment variable name is invalid")
|
|
if clean_name not in _allowed_secret_env_names():
|
|
raise ConnectorDeploymentConfigurationError(
|
|
f"Connector credential environment variable is not listed in {_SECRET_ENV_ALLOWLIST}"
|
|
)
|
|
return clean_name
|
|
|
|
|
|
def _allowed_secret_env_names() -> frozenset[str]:
|
|
raw = os.environ.get(_SECRET_ENV_ALLOWLIST, "")
|
|
names = frozenset(item.strip() for item in raw.split(",") if item.strip())
|
|
invalid = sorted(name for name in names if not _ENV_NAME.fullmatch(name))
|
|
if invalid:
|
|
raise ConnectorDeploymentConfigurationError(f"{_SECRET_ENV_ALLOWLIST} contains an invalid variable name")
|
|
return names
|
|
|
|
|
|
def _allowed_ca_bundle_paths() -> frozenset[Path]:
|
|
raw = os.environ.get(_CA_BUNDLE_ALLOWLIST, "")
|
|
paths: set[Path] = set()
|
|
for item in (part.strip() for part in raw.split(",")):
|
|
if not item:
|
|
continue
|
|
path = Path(item)
|
|
if not path.is_absolute():
|
|
raise ConnectorDeploymentConfigurationError(f"{_CA_BUNDLE_ALLOWLIST} requires absolute paths")
|
|
try:
|
|
paths.add(path.resolve(strict=True))
|
|
except OSError as exc:
|
|
raise ConnectorDeploymentConfigurationError(f"{_CA_BUNDLE_ALLOWLIST} contains a missing path") from exc
|
|
return frozenset(paths)
|
|
|
|
|
|
def _development_runtime() -> bool:
|
|
app_env = os.environ.get("APP_ENV", "dev").strip().casefold()
|
|
install_profile = os.environ.get("GOVOPLAN_INSTALL_PROFILE", "").strip().casefold()
|
|
return app_env in _DEVELOPMENT_ENVIRONMENTS and (
|
|
not install_profile or install_profile in _DEVELOPMENT_ENVIRONMENTS
|
|
)
|
|
|
|
|
|
def _as_bool(value: object, *, default: bool) -> bool:
|
|
if value is None:
|
|
return default
|
|
if isinstance(value, bool):
|
|
return value
|
|
clean = str(value).strip().casefold()
|
|
if clean in {"1", "true", "yes", "on"}:
|
|
return True
|
|
if clean in {"0", "false", "no", "off"}:
|
|
return False
|
|
raise ConnectorDeploymentConfigurationError("Connector TLS verification setting must be true or false")
|
|
|
|
|
|
def _clean(value: object) -> str | None:
|
|
if value is None:
|
|
return None
|
|
clean = str(value).strip()
|
|
return clean or None
|