GovOPlaN/govoplan-core
1
0
Fork 0
Code Issues 43 Pull Requests Actions Packages Projects Releases Wiki Activity

Harden module installation and imports

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-11 18:37:51 +02:00
parent 9a0c467d55
commit b9badc9153
13 changed files with 194 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/govoplan_core/core/configuration_packages.py
View File

@@ -690,7 +690,7 @@ def _read_trusted_keys_url(url: str) -> str:
raise ValueError("Trusted configuration catalog key URL must use http:// or https://.")
cache_path = _configured_trusted_keys_cache_path()
try:
with urllib.request.urlopen(url, timeout=15) as response:
with urllib.request.urlopen(url, timeout=15) as response: # noqa: S310 - validated configuration key HTTP(S) URL. # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
body = response.read().decode("utf-8")
if cache_path is not None:
cache_path.parent.mkdir(parents=True, exist_ok=True)
@@ -714,7 +714,7 @@ def _read_catalog_payload_with_metadata(source: Path | str | None) -> tuple[obje
return {"packages": []}, metadata
if isinstance(source, str) and _is_http_url(source):
try:
with urllib.request.urlopen(source, timeout=15) as response:
with urllib.request.urlopen(source, timeout=15) as response: # noqa: S310 - validated configuration catalog HTTP(S) URL. # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
body = response.read().decode("utf-8")
if cache_path is not None:
cache_path.parent.mkdir(parents=True, exist_ok=True)

64
src/govoplan_core/core/module_installer.py
View File

@@ -11,6 +11,7 @@ import json
import os
from pathlib import Path
import re
import shlex
import shutil
import sqlite3
import subprocess
@@ -2611,7 +2612,7 @@ def _run_restart_command(command: str) -> dict[str, object]:
"stdout": "",
"stderr": "",
}
result = subprocess.run(command, shell=True, text=True, capture_output=True, check=False)
result = _run_operator_command(command)
return {
"command": command,
"return_code": result.returncode,
@@ -2685,13 +2686,16 @@ def _python_package_name_from_ref(value: str) -> str | None:
def _wait_for_health_url(url: str, *, timeout_seconds: float, interval_seconds: float) -> dict[str, object]:
parsed = urllib.parse.urlparse(url)
if parsed.scheme not in {"http", "https"} or not parsed.netloc or parsed.username or parsed.password:
return {"ok": False, "attempts": 0, "error": "health URL must be an absolute HTTP(S) URL without embedded credentials"}
deadline = time.monotonic() + max(timeout_seconds, 0.1)
attempts = 0
last_error = ""
while time.monotonic() < deadline:
attempts += 1
try:
with urllib.request.urlopen(url, timeout=min(max(interval_seconds, 0.5), 10.0)) as response:
with urllib.request.urlopen(url, timeout=min(max(interval_seconds, 0.5), 10.0)) as response: # noqa: S310 - validated module health HTTP(S) URL. # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
status = int(getattr(response, "status", 0))
if 200 <= status < 400:
return {"ok": True, "status": status, "attempts": attempts}
@@ -3217,15 +3221,53 @@ def _run_database_hook(
if pgtools_url:
env["GOVOPLAN_DATABASE_URL_PGTOOLS"] = pgtools_url
env.setdefault("DATABASE_URL", database_url)
return subprocess.run(
command,
cwd=run_dir,
env=env,
shell=True,
text=True,
capture_output=True,
check=False,
)
return _run_operator_command(command, cwd=run_dir, env=env)
def _run_operator_command(
command: str,
*,
cwd: Path | None = None,
env: Mapping[str, str] | None = None,
) -> subprocess.CompletedProcess[str]:
try:
argv = _operator_command_argv(command)
except ValueError as exc:
return subprocess.CompletedProcess(args=command, returncode=2, stdout="", stderr=str(exc))
try:
return subprocess.run(
argv,
cwd=cwd,
env=dict(env) if env is not None else None,
text=True,
capture_output=True,
check=False,
)
except OSError as exc:
return subprocess.CompletedProcess(args=argv, returncode=127, stdout="", stderr=str(exc))
def _operator_command_argv(command: str) -> tuple[str, ...]:
try:
argv = tuple(shlex.split(command))
except ValueError as exc:
raise ValueError(f"Invalid command syntax: {exc}") from exc
if not argv:
raise ValueError("Command is empty")
if any(_operator_command_part_uses_shell(part) for part in argv):
raise ValueError("Shell operators, substitutions, and redirects are not supported in installer hooks")
if _looks_like_env_assignment(argv[0]):
raise ValueError("Environment assignments must be configured through installer hook environment, not shell syntax")
return argv
def _operator_command_part_uses_shell(value: str) -> bool:
return value in {"|", "||", "&", "&&", ";", "<", ">", ">>", "2>", "2>>"} or "$(" in value or "`" in value
def _looks_like_env_assignment(value: str) -> bool:
key, separator, _rest = value.partition("=")
return bool(separator) and bool(re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", key))
def _database_url_for_pgtools(database_url: str) -> str | None:

8
src/govoplan_core/core/module_license.py
View File

@@ -8,6 +8,7 @@ import os
from pathlib import Path
from typing import Any
import urllib.error
import urllib.parse
import urllib.request
from cryptography.exceptions import InvalidSignature
@@ -257,11 +258,12 @@ def _configured_trusted_keys_cache_path() -> Path | None:
def _read_trusted_keys_url(url: str) -> str:
if not url.startswith(("https://", "http://")):
raise ValueError("Trusted license key URL must use http:// or https://.")
parsed = urllib.parse.urlparse(url)
if parsed.scheme not in {"http", "https"} or not parsed.netloc or parsed.username or parsed.password:
raise ValueError("Trusted license key URL must be an absolute HTTP(S) URL without embedded credentials.")
cache_path = _configured_trusted_keys_cache_path()
try:
with urllib.request.urlopen(url, timeout=15) as response:
with urllib.request.urlopen(url, timeout=15) as response: # noqa: S310 - validated license key HTTP(S) URL. # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
body = response.read().decode("utf-8")
if cache_path is not None:
cache_path.parent.mkdir(parents=True, exist_ok=True)

7
src/govoplan_core/core/module_package_catalog.py
View File

@@ -354,7 +354,7 @@ def _read_trusted_keys_url(url: str) -> str:
raise ValueError("Trusted catalog key URL must use http:// or https://.")
cache_path = _configured_trusted_keys_cache_path()
try:
with urllib.request.urlopen(url, timeout=15) as response:
with urllib.request.urlopen(url, timeout=15) as response: # noqa: S310 - validated catalog key HTTP(S) URL. # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
body = response.read().decode("utf-8")
if cache_path is not None:
cache_path.parent.mkdir(parents=True, exist_ok=True)
@@ -421,7 +421,7 @@ def _read_catalog_url(url: str) -> str:
def _read_catalog_url_with_metadata(url: str) -> tuple[str, bool]:
cache_path = _configured_catalog_cache_path()
try:
with urllib.request.urlopen(url, timeout=15) as response:
with urllib.request.urlopen(url, timeout=15) as response: # noqa: S310 - validated module catalog HTTP(S) URL. # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
body = response.read().decode("utf-8")
if cache_path is not None:
cache_path.parent.mkdir(parents=True, exist_ok=True)
@@ -935,7 +935,8 @@ def _catalog_source_type(source: Path | str | None) -> str | None:
def _is_http_url(value: str) -> bool:
return value.startswith(("https://", "http://"))
parsed = urllib.parse.urlparse(value)
return parsed.scheme in {"http", "https"} and bool(parsed.netloc) and not parsed.username and not parsed.password
def _invalid_catalog_result(