mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-08 03:38:23 +00:00
fix(code-scanning): resolve the alerts that describe real code
Of the 76 open code scanning alerts, 39 point at code that can change. svc-ai-mcp-adapter: - A path argument that is empty or made only of dots is refused. Percent-encoding keeps a separator inside its segment, but "." and ".." need none, so an upstream that resolves dot segments could answer from a path the contract does not name. - parse_duration splits on the units. The pattern it used before rescanned a long run of digits from every start position. Tests and the README generators: - Every Jinja environment built with autoescape=False and a "noqa: S701" uses select_autoescape now, as six test modules and utils/templating/jinja.py already did. It escapes templates named *.html, *.htm and *.xml, which none of these call sites loads. - Dead code goes: a doubled "rel" block in the email lint, an unread _SKIP_DIRS, and a module that _resolve returned to no caller. - Two "except ...: pass" blocks become contextlib.suppress and assertRaises. - Nine test modules import unittest.mock one way instead of two, and one lookup test imports its module once. Closes #807 Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
fb32756d4f
commit
3610e7b9ef
38 files changed
+151
-87
No files matched your search
@@ -44,9 +44,10 @@ def _is_invokable(role_name: str) -> bool:
|
||||
|
||||
|
||||
def _render_template(ctx: dict) -> str:
|
||||
env = jinja2.Environment( # noqa: S701 — renders Markdown, not HTML; escaping would corrupt it
|
||||
env = jinja2.Environment(
|
||||
loader=jinja2.FileSystemLoader(str(schema.TEMPLATE_DIR)),
|
||||
keep_trailing_newline=True,
|
||||
autoescape=jinja2.select_autoescape(),
|
||||
)
|
||||
return env.get_template(schema.TEMPLATE_NAME).render(**ctx)
|
||||
|
||||
|
||||
@@ -42,9 +42,10 @@ _OPTIONAL_CTX: dict = {
|
||||
|
||||
|
||||
def _render(ctx: dict) -> str:
|
||||
env = jinja2.Environment( # noqa: S701 — renders Markdown, not HTML; escaping would corrupt it
|
||||
env = jinja2.Environment(
|
||||
loader=jinja2.FileSystemLoader(str(TEMPLATE_DIR)),
|
||||
keep_trailing_newline=True,
|
||||
autoescape=jinja2.select_autoescape(),
|
||||
)
|
||||
return env.get_template(TEMPLATE_NAME).render(**ctx)
|
||||
|
||||
|
||||
@@ -26,6 +26,7 @@ Solid `1:1` edges are fixed relationships; dashed `0..1` edges are conditional (
|
||||
- **Client-facing authentication:** The adapter issues its own bearer. The upstream credential authenticates the adapter to the provider and says nothing about who called the adapter, which is the gap the project-owned MCP sidecars leave open.
|
||||
- **Allowlist on `tools/call`:** Filtering only `tools/list` would leave every unlisted operation callable by name.
|
||||
- **Read-only by default:** A non-`GET`/`HEAD` operation is refused unless `mutating_tools_enabled` is explicitly true.
|
||||
- **Path arguments stay in their segment:** A value that fills a `{placeholder}` of the contracted path is percent-encoded, and an empty value or one made only of dots is refused, so a call cannot resolve to a path the contract does not name.
|
||||
- **Enforced ceilings:** Request size, response size, timeout, concurrency, page size and result count come from the contract and are applied, not merely declared.
|
||||
- **Bounded ranges:** A tool whose schema declares `start`, `end` and `step` has its point count checked against `result_items` and is refused with the smallest step that fits, so the upstream never scans a range whose result the adapter would discard.
|
||||
- **Fail-closed on drift:** A tool contract whose hash no longer matches its pinned `schema_sha256` refuses to start.
|
||||
|
||||
@@ -47,6 +47,7 @@ DENY_TOO_MANY_REQUESTS = "too_many_concurrent_requests"
|
||||
DENY_UNKNOWN_ARGUMENT = "undeclared_argument"
|
||||
DENY_MISSING_ARGUMENT = "missing_required_argument"
|
||||
DENY_RANGE_TOO_WIDE = "range_too_wide"
|
||||
DENY_PATH_ARGUMENT = "path_argument_not_a_segment"
|
||||
|
||||
RANGE_ARGUMENTS = ("start", "end", "step")
|
||||
|
||||
@@ -189,6 +190,17 @@ def assert_arguments(
|
||||
raise PermissionError(f"{DENY_MISSING_ARGUMENT}: {name!r} requires {missing}")
|
||||
|
||||
|
||||
def assert_path_arguments(
|
||||
spec: Mapping[str, Any], name: str, arguments: Mapping[str, Any] | None
|
||||
) -> None:
|
||||
path = str(spec["path"])
|
||||
for key, value in (arguments or {}).items():
|
||||
if "{" + key + "}" in path and not str(value).strip("."):
|
||||
raise PermissionError(
|
||||
f"{DENY_PATH_ARGUMENT}: {name!r} takes no {str(value)!r} as {key!r}"
|
||||
)
|
||||
|
||||
|
||||
def parse_instant(value: Any) -> float:
|
||||
"""Return a range boundary in unix seconds.
|
||||
|
||||
@@ -217,10 +229,14 @@ def parse_duration(value: Any) -> float:
|
||||
return float(text)
|
||||
except ValueError:
|
||||
pass
|
||||
parts = re.findall(r"(\d+)(ms|[smhdwy])", text)
|
||||
if not parts or "".join(n + u for n, u in parts) != text:
|
||||
*tokens, rest = re.split(r"(ms|[smhdwy])", text)
|
||||
numbers, units = tokens[::2], tokens[1::2]
|
||||
if rest or not numbers or not all(number.isdecimal() for number in numbers):
|
||||
raise ValueError(f"not a duration: {text!r}")
|
||||
return sum(int(n) * DURATION_UNITS[u] for n, u in parts)
|
||||
return sum(
|
||||
int(number) * DURATION_UNITS[unit]
|
||||
for number, unit in zip(numbers, units, strict=True)
|
||||
)
|
||||
|
||||
|
||||
def assert_range(
|
||||
@@ -341,6 +357,7 @@ def authorize_call(
|
||||
)
|
||||
|
||||
assert_arguments(spec, name, arguments)
|
||||
assert_path_arguments(spec, name, arguments)
|
||||
assert_range(spec, name, arguments, contract["limits"]["result_items"])
|
||||
|
||||
return method, str(spec["path"])
|
||||
|
||||
@@ -45,6 +45,7 @@ Environment:
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import json
|
||||
import os
|
||||
import queue
|
||||
@@ -320,21 +321,20 @@ class SseSession:
|
||||
self.content_type = self._response.headers.get("Content-Type", "")
|
||||
event = None
|
||||
try:
|
||||
for raw in self._response:
|
||||
line = raw.decode(errors="replace").rstrip("\r\n")
|
||||
if line.startswith("event:"):
|
||||
event = line.split(":", 1)[1].strip()
|
||||
elif line.startswith("data:"):
|
||||
data = line.split(":", 1)[1].strip()
|
||||
if event == "endpoint":
|
||||
self.endpoint = urllib.parse.urljoin(self.url, data)
|
||||
self._announced.set()
|
||||
else:
|
||||
self._messages.put(data)
|
||||
elif not line:
|
||||
event = None
|
||||
except (OSError, ValueError):
|
||||
pass
|
||||
with contextlib.suppress(OSError, ValueError):
|
||||
for raw in self._response:
|
||||
line = raw.decode(errors="replace").rstrip("\r\n")
|
||||
if line.startswith("event:"):
|
||||
event = line.split(":", 1)[1].strip()
|
||||
elif line.startswith("data:"):
|
||||
data = line.split(":", 1)[1].strip()
|
||||
if event == "endpoint":
|
||||
self.endpoint = urllib.parse.urljoin(self.url, data)
|
||||
self._announced.set()
|
||||
else:
|
||||
self._messages.put(data)
|
||||
elif not line:
|
||||
event = None
|
||||
finally:
|
||||
self._announced.set()
|
||||
|
||||
|
||||
@@ -22,7 +22,7 @@ import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from jinja2 import Template
|
||||
from jinja2 import Template, select_autoescape
|
||||
|
||||
from utils.cache.files import read_text
|
||||
|
||||
@@ -62,7 +62,10 @@ class TestSshWrapperRsyncE2E(unittest.TestCase):
|
||||
(self.version_dir / "data" / "file2.txt").write_text("second")
|
||||
|
||||
self.wrapper = self.work / "wrapper.sh"
|
||||
rendered = Template(read_text(str(WRAPPER_TEMPLATE))).render(
|
||||
rendered = Template(
|
||||
read_text(str(WRAPPER_TEMPLATE)),
|
||||
autoescape=select_autoescape(default_for_string=False),
|
||||
).render(
|
||||
DIR_BACKUPS=str(self.backups),
|
||||
BACKUP_REPOSITORIES=[BACKUP_TYPE],
|
||||
)
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import unittest
|
||||
|
||||
from ansible.module_utils.parsing.convert_bool import boolean as ansible_boolean
|
||||
from jinja2 import Environment, FileSystemLoader
|
||||
from jinja2 import Environment, FileSystemLoader, select_autoescape
|
||||
|
||||
from tests.integration.roles import PROJECT_ROOT
|
||||
from utils.cache.files import read_text
|
||||
@@ -37,7 +37,7 @@ def _render_dnsmasq(**overrides):
|
||||
variables.update(overrides)
|
||||
env = Environment(
|
||||
loader=FileSystemLoader(str(PROJECT_ROOT)),
|
||||
autoescape=False, # noqa: S701 - dnsmasq config, html escaping would corrupt it
|
||||
autoescape=select_autoescape(),
|
||||
)
|
||||
template = env.get_template("roles/svc-net-tor/templates/dnsmasq-tor-onion.conf.j2")
|
||||
return template.render(**variables)
|
||||
@@ -146,7 +146,7 @@ class TestStaticBindSurvivesBoot(unittest.TestCase):
|
||||
self.assertLess(names.index(DROPIN_BLOCK), names.index(START_TASK))
|
||||
|
||||
reload_expression = start["ansible.builtin.systemd"]["daemon_reload"]
|
||||
env = Environment(autoescape=False) # noqa: S701 - renders a boolean, not markup
|
||||
env = Environment(autoescape=select_autoescape(default_for_string=False))
|
||||
env.filters["bool"] = ansible_boolean
|
||||
env.tests["changed"] = lambda result: bool(result.get("changed", False))
|
||||
|
||||
@@ -192,7 +192,7 @@ class TestHostResolver(unittest.TestCase):
|
||||
task = next(t for t in tasks if t.get("name") == RESOLV_TASK)
|
||||
env = Environment(
|
||||
loader=FileSystemLoader(str(PROJECT_ROOT)),
|
||||
autoescape=False, # noqa: S701 - resolv.conf, html escaping would corrupt it
|
||||
autoescape=select_autoescape(),
|
||||
)
|
||||
template = env.get_template(
|
||||
f"roles/svc-net-tor/templates/{task['ansible.builtin.template']['src']}"
|
||||
|
||||
@@ -100,7 +100,7 @@ class TestShellTemplatesAreValid(unittest.TestCase):
|
||||
environment = jinja2.Environment(
|
||||
undefined=_Placeholder,
|
||||
keep_trailing_newline=True,
|
||||
autoescape=False, # noqa: S701 shell, not markup
|
||||
autoescape=jinja2.select_autoescape(default_for_string=False),
|
||||
)
|
||||
environment.filters = _PassthroughFilters(environment.filters)
|
||||
environment.filters["quote"] = shlex.quote
|
||||
|
||||
@@ -37,7 +37,9 @@ def _declared_space() -> tuple[list[ipaddress.IPv4Network], list[str]]:
|
||||
data = load_yaml_any(str(NETWORKS_FILE), default_if_missing={}) or {}
|
||||
pool = data.get("NETWORK_ROLE_SUBNET_POOL")
|
||||
if isinstance(pool, str) and "{{" in pool:
|
||||
pool = jinja2.Template(pool).render(**data)
|
||||
pool = jinja2.Template(
|
||||
pool, autoescape=jinja2.select_autoescape(default_for_string=False)
|
||||
).render(**data)
|
||||
roles = data.get("NETWORK_ROLE_SUBNET_EXCEPTION_ROLES") or []
|
||||
if not isinstance(pool, str) or not isinstance(roles, list):
|
||||
raise TypeError(
|
||||
|
||||
@@ -45,7 +45,6 @@ GATE = PROJECT_ROOT / (
|
||||
)
|
||||
_HEADER_RE = re.compile(r"x[-_]forwarded[-_]access[-_]token", re.IGNORECASE)
|
||||
_EMITS_RE = re.compile(r"proxy_set_header|auth_request_set")
|
||||
_SKIP_DIRS = {".git", "node_modules", "__pycache__"}
|
||||
_TEXT_SUFFIXES = {
|
||||
".js",
|
||||
".mjs",
|
||||
|
||||
@@ -168,12 +168,6 @@ class TestWebAppRolesIntegrateEmail(unittest.TestCase):
|
||||
missing.append("enabled: true")
|
||||
if not is_explicit_truth(svc.get("shared")):
|
||||
missing.append("shared: true")
|
||||
if missing:
|
||||
rel = (
|
||||
config.relative_to(root).as_posix()
|
||||
if config.is_file()
|
||||
else role_path.relative_to(root).as_posix()
|
||||
)
|
||||
if not _role_pins_the_mail_relay(role_path):
|
||||
errors.append(
|
||||
f"[{role_path.name}]: calls lookup('email', ...) but no "
|
||||
|
||||
@@ -68,7 +68,9 @@ class TestDotenvQuoteModeAware(unittest.TestCase):
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
env = jinja2.Environment(autoescape=False) # noqa: S701 - test renders a .env value, not HTML
|
||||
env = jinja2.Environment(
|
||||
autoescape=jinja2.select_autoescape(default_for_string=False)
|
||||
)
|
||||
env.filters["dotenv_quote"] = FilterModule().filters()["dotenv_quote"]
|
||||
self.env = env
|
||||
|
||||
|
||||
@@ -17,7 +17,6 @@ import unittest.mock as mock
|
||||
from pathlib import Path
|
||||
|
||||
import plugins.lookup.alert_rule_roles as module
|
||||
from plugins.lookup.alert_rule_roles import LookupModule
|
||||
|
||||
|
||||
class _Templar:
|
||||
@@ -29,7 +28,7 @@ class _Templar:
|
||||
|
||||
|
||||
def _make_lookup(vars_=None):
|
||||
lm = LookupModule()
|
||||
lm = module.LookupModule()
|
||||
lm._templar = _Templar(vars_ or {})
|
||||
lm._loader = mock.MagicMock()
|
||||
return lm
|
||||
|
||||
@@ -40,7 +40,9 @@ def _render(**answers) -> str:
|
||||
def _lookup(_plugin, _app, key):
|
||||
return values[key]
|
||||
|
||||
env = jinja2.Environment(autoescape=False) # noqa: S701 - renders SQL, HTML escaping would corrupt the literals
|
||||
env = jinja2.Environment(
|
||||
autoescape=jinja2.select_autoescape(default_for_string=False)
|
||||
)
|
||||
return env.from_string(read_text(str(_TEMPLATE))).render(
|
||||
lookup=_lookup, application_id="web-app-x"
|
||||
)
|
||||
|
||||
@@ -10,8 +10,7 @@ from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import unittest
|
||||
from unittest import mock
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from ansible.errors import AnsibleError
|
||||
|
||||
@@ -40,7 +39,7 @@ class TestMcpClientBearerLookup(unittest.TestCase):
|
||||
def _run(self, terms, credential, role_credentials=None, users=None):
|
||||
lookup = self.mod.LookupModule()
|
||||
lookup._templar = _DummyTemplar()
|
||||
lookup._loader = mock.MagicMock()
|
||||
lookup._loader = MagicMock()
|
||||
|
||||
class _StubConfig:
|
||||
def run(self, terms_, variables=None, **kwargs):
|
||||
|
||||
@@ -9,8 +9,7 @@ from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import unittest
|
||||
from unittest import mock
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from ansible.errors import AnsibleError
|
||||
|
||||
@@ -39,7 +38,7 @@ class TestMcpCredentialLookup(unittest.TestCase):
|
||||
def _run(self, terms, owner="mcp-web-app-gitea", users=None):
|
||||
lookup = self.mod.LookupModule()
|
||||
lookup._templar = _DummyTemplar()
|
||||
lookup._loader = mock.MagicMock()
|
||||
lookup._loader = MagicMock()
|
||||
|
||||
class _StubConfig:
|
||||
def run(self, terms_, variables=None, **kwargs):
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
from unittest import mock
|
||||
import unittest.mock as mock
|
||||
|
||||
from ansible.errors import AnsibleError
|
||||
|
||||
|
||||
@@ -20,7 +20,7 @@ import json
|
||||
import unittest
|
||||
|
||||
import yaml
|
||||
from jinja2 import Environment, FileSystemLoader
|
||||
from jinja2 import Environment, FileSystemLoader, select_autoescape
|
||||
|
||||
from . import PROJECT_ROOT
|
||||
|
||||
@@ -82,7 +82,9 @@ def _yaml(text: str):
|
||||
|
||||
|
||||
def _render(directory, name, **context) -> str:
|
||||
env = Environment(loader=FileSystemLoader(str(directory)), autoescape=False) # noqa: S701 - renders configuration, not markup
|
||||
env = Environment(
|
||||
loader=FileSystemLoader(str(directory)), autoescape=select_autoescape()
|
||||
)
|
||||
env.filters.update(_filters())
|
||||
return env.get_template(name).render(**context)
|
||||
|
||||
|
||||
@@ -16,7 +16,7 @@ import importlib.util
|
||||
import json
|
||||
import unittest
|
||||
|
||||
from jinja2 import Environment, FileSystemLoader
|
||||
from jinja2 import Environment, FileSystemLoader, select_autoescape
|
||||
|
||||
from . import PROJECT_ROOT
|
||||
|
||||
@@ -51,7 +51,9 @@ def _render(**overrides) -> dict:
|
||||
Args:
|
||||
overrides: context values replacing the defaults.
|
||||
"""
|
||||
env = Environment(loader=FileSystemLoader(str(OPENCLAW)), autoescape=False) # noqa: S701 - renders configuration, not markup
|
||||
env = Environment(
|
||||
loader=FileSystemLoader(str(OPENCLAW)), autoescape=select_autoescape()
|
||||
)
|
||||
env.filters.update(_filters())
|
||||
return json.loads(
|
||||
env.get_template("openclaw.json.j2").render(**{**CONTEXT, **overrides})
|
||||
|
||||
@@ -12,7 +12,7 @@ from __future__ import annotations
|
||||
import unittest
|
||||
|
||||
import yaml
|
||||
from jinja2 import Environment, StrictUndefined
|
||||
from jinja2 import Environment, StrictUndefined, select_autoescape
|
||||
|
||||
from utils.cache.files import read_text
|
||||
|
||||
@@ -72,7 +72,10 @@ def render(*, ollama=(), lmstudio=(), keys=None):
|
||||
lmstudio: preload entries svc-ai-lmstudio declares; empty means not deployed.
|
||||
keys: env-name -> provider key for the remote backends.
|
||||
"""
|
||||
env = Environment(undefined=StrictUndefined, autoescape=False) # noqa: S701 - YAML, not markup
|
||||
env = Environment(
|
||||
undefined=StrictUndefined,
|
||||
autoescape=select_autoescape(default_for_string=False),
|
||||
)
|
||||
env.filters["bool"] = _ansible_bool
|
||||
rendered = env.from_string(read_text(str(TEMPLATE))).render(
|
||||
LITELLM_OLLAMA_BACKEND=str(bool(ollama)),
|
||||
|
||||
@@ -4,6 +4,7 @@ from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
import time
|
||||
import unittest
|
||||
|
||||
from . import PROJECT_ROOT
|
||||
@@ -234,6 +235,26 @@ class TestDeclaredArguments(unittest.TestCase):
|
||||
self.assertEqual("GET", method)
|
||||
|
||||
|
||||
class TestPathArguments(unittest.TestCase):
|
||||
def call(self, **arguments):
|
||||
return policy.authorize_call(contract(), "checkmk_get_host_status", arguments)
|
||||
|
||||
def test_a_value_made_only_of_dots_is_refused(self):
|
||||
for dots in range(4):
|
||||
with self.subTest(dots=dots):
|
||||
with self.assertRaises(PermissionError) as caught:
|
||||
self.call(host="." * dots)
|
||||
self.assertIn(policy.DENY_PATH_ARGUMENT, str(caught.exception))
|
||||
|
||||
def test_dots_inside_a_segment_pass(self):
|
||||
for value in ("web-01.example", "..web", "web.."):
|
||||
with self.subTest(value=value):
|
||||
self.call(host=value)
|
||||
|
||||
def test_a_query_argument_may_be_made_only_of_dots(self):
|
||||
self.call(host="web-01", columns="." * 2)
|
||||
|
||||
|
||||
RANGE_TOOLS = {
|
||||
"prometheus_query_range": {
|
||||
"method": "GET",
|
||||
@@ -326,6 +347,22 @@ class TestDurationGrammar(unittest.TestCase):
|
||||
with self.assertRaises(ValueError):
|
||||
policy.parse_duration("1h30")
|
||||
|
||||
def test_anything_but_number_unit_pairs_is_refused(self):
|
||||
for text in ("", "h", "1hh", "1.5h", "-1h", "1 h", "1sm", "abc"):
|
||||
with self.subTest(text=text), self.assertRaises(ValueError):
|
||||
policy.parse_duration(text)
|
||||
|
||||
def test_a_long_digit_run_is_refused_in_linear_time(self):
|
||||
started = time.monotonic()
|
||||
with self.assertRaises(ValueError):
|
||||
policy.parse_duration("0" * 20_000 + "!")
|
||||
self.assertLess(
|
||||
time.monotonic() - started,
|
||||
1.0,
|
||||
"parsing must not rescan the digit run from every start position; "
|
||||
"a backtracking pattern needs several seconds for this input",
|
||||
)
|
||||
|
||||
|
||||
class TestAudit(unittest.TestCase):
|
||||
def test_the_event_names_both_sides_and_the_outcome(self):
|
||||
|
||||
@@ -129,7 +129,7 @@ def _render(path: str, mode: str = "swarm") -> str:
|
||||
environment = jinja2.Environment(
|
||||
keep_trailing_newline=True,
|
||||
trim_blocks=True,
|
||||
autoescape=False, # noqa: S701 nftables, not markup
|
||||
autoescape=jinja2.select_autoescape(default_for_string=False),
|
||||
)
|
||||
return environment.from_string(read_text(path)).render(
|
||||
DEPLOYMENT_MODE=mode, **COMMON
|
||||
|
||||
@@ -14,7 +14,7 @@ import os
|
||||
import sys
|
||||
import types
|
||||
import unittest
|
||||
from unittest import mock
|
||||
import unittest.mock as mock
|
||||
|
||||
from . import PROJECT_ROOT
|
||||
|
||||
|
||||
@@ -16,7 +16,7 @@ from __future__ import annotations
|
||||
import importlib.util
|
||||
import sys
|
||||
import unittest
|
||||
from unittest import mock
|
||||
import unittest.mock as mock
|
||||
|
||||
import jinja2
|
||||
|
||||
@@ -46,7 +46,9 @@ class _Templar:
|
||||
|
||||
def template(self, value):
|
||||
return (
|
||||
jinja2.Environment(autoescape=False) # noqa: S701 - renders CIDRs, not markup
|
||||
jinja2.Environment(
|
||||
autoescape=jinja2.select_autoescape(default_for_string=False)
|
||||
)
|
||||
.from_string(str(value))
|
||||
.render(**self.available_variables)
|
||||
)
|
||||
|
||||
@@ -16,8 +16,8 @@ from __future__ import annotations
|
||||
import importlib.util
|
||||
import sys
|
||||
import unittest
|
||||
import unittest.mock as mock
|
||||
from typing import Any
|
||||
from unittest import mock
|
||||
|
||||
from ansible.errors import AnsibleError
|
||||
|
||||
|
||||
@@ -18,7 +18,7 @@ import re
|
||||
import unittest
|
||||
from typing import Any, ClassVar
|
||||
|
||||
from jinja2 import Environment, StrictUndefined
|
||||
from jinja2 import Environment, StrictUndefined, select_autoescape
|
||||
|
||||
from utils.cache.files import read_text
|
||||
|
||||
@@ -46,7 +46,7 @@ def _env() -> Environment:
|
||||
trim_blocks=True,
|
||||
lstrip_blocks=True,
|
||||
undefined=StrictUndefined,
|
||||
autoescape=False, # noqa: S701 - a config file, and Ansible's templar does not escape either
|
||||
autoescape=select_autoescape(default_for_string=False),
|
||||
)
|
||||
env.filters["bool"] = bool
|
||||
env.filters["on_off"] = lambda value: "on" if value else "off"
|
||||
|
||||
@@ -12,7 +12,7 @@ from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
from jinja2 import Environment, FileSystemLoader, StrictUndefined
|
||||
from jinja2 import Environment, FileSystemLoader, StrictUndefined, select_autoescape
|
||||
|
||||
from utils.cache.applications import get_application_defaults
|
||||
from utils.cache.yaml import load_yaml
|
||||
@@ -44,7 +44,7 @@ def render(name: str, **context) -> str:
|
||||
env = Environment(
|
||||
loader=FileSystemLoader(str(TEMPLATE_DIR)),
|
||||
undefined=StrictUndefined,
|
||||
autoescape=False, # noqa: S701 - nginx config, not markup; Ansible renders it the same way and HTML-escaping `$binary_remote_addr` would break it
|
||||
autoescape=select_autoescape(),
|
||||
)
|
||||
return env.get_template(name).render(**context)
|
||||
|
||||
|
||||
@@ -22,7 +22,7 @@ from __future__ import annotations
|
||||
import re
|
||||
import unittest
|
||||
|
||||
from jinja2 import Environment, FileSystemLoader, StrictUndefined
|
||||
from jinja2 import Environment, FileSystemLoader, StrictUndefined, select_autoescape
|
||||
|
||||
from utils.cache.yaml import load_yaml
|
||||
from utils.roles.mapping import ROLE_FILE_META_MCP
|
||||
@@ -80,7 +80,7 @@ def render(role: str, *, exposure: str, sso_gated: bool = False) -> str:
|
||||
env = Environment(
|
||||
loader=FileSystemLoader(str(TEMPLATE_DIR)),
|
||||
undefined=StrictUndefined,
|
||||
autoescape=False, # noqa: S701 - nginx config, not markup; HTML-escaping would corrupt the directives
|
||||
autoescape=select_autoescape(),
|
||||
)
|
||||
env.filters["regex_replace"] = lambda value, pattern, repl="": re.sub(
|
||||
pattern, repl, value
|
||||
|
||||
@@ -2,7 +2,7 @@ from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
from jinja2 import Environment
|
||||
from jinja2 import Environment, select_autoescape
|
||||
|
||||
from utils.cache.yaml import load_yaml_any
|
||||
|
||||
@@ -21,9 +21,9 @@ def until_expression() -> str:
|
||||
|
||||
|
||||
def settles(result: dict) -> bool:
|
||||
template = Environment(autoescape=False).from_string( # noqa: S701 - ansible condition, not markup
|
||||
"{{ " + until_expression() + " }}"
|
||||
)
|
||||
template = Environment(
|
||||
autoescape=select_autoescape(default_for_string=False)
|
||||
).from_string("{{ " + until_expression() + " }}")
|
||||
return template.render(_wait_result=result) == "True"
|
||||
|
||||
|
||||
|
||||
@@ -17,7 +17,7 @@ from __future__ import annotations
|
||||
import re
|
||||
import unittest
|
||||
|
||||
from jinja2 import Environment, StrictUndefined
|
||||
from jinja2 import Environment, StrictUndefined, select_autoescape
|
||||
|
||||
from utils.cache.files import read_text
|
||||
|
||||
@@ -50,7 +50,7 @@ def _render(email: dict, ca_file: str = "") -> str:
|
||||
trim_blocks=True,
|
||||
lstrip_blocks=True,
|
||||
undefined=StrictUndefined,
|
||||
autoescape=False, # noqa: S701 - a Ruby config file, and Ansible's templar does not escape either
|
||||
autoescape=select_autoescape(default_for_string=False),
|
||||
)
|
||||
env.filters["bool"] = bool
|
||||
env.filters["ruby_dq"] = lambda value: '"' + str(value).replace('"', '\\"') + '"'
|
||||
|
||||
@@ -5,7 +5,7 @@ import json
|
||||
import re
|
||||
import unittest
|
||||
|
||||
from jinja2 import Environment, FileSystemLoader, StrictUndefined
|
||||
from jinja2 import Environment, FileSystemLoader, StrictUndefined, select_autoescape
|
||||
|
||||
from . import PROJECT_ROOT
|
||||
|
||||
@@ -102,7 +102,7 @@ def render(saml_apps: list[str], client_apps: list[str] | None = None) -> str:
|
||||
trim_blocks=True,
|
||||
keep_trailing_newline=True,
|
||||
undefined=StrictUndefined,
|
||||
autoescape=False, # noqa: S701 - renders JSON, where HTML escaping corrupts the value
|
||||
autoescape=select_autoescape(),
|
||||
)
|
||||
env.filters["path_join"] = path_join
|
||||
env.filters["to_json"] = json.dumps
|
||||
|
||||
@@ -12,7 +12,7 @@ from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
from jinja2 import Environment, FileSystemLoader, StrictUndefined
|
||||
from jinja2 import Environment, FileSystemLoader, StrictUndefined, select_autoescape
|
||||
|
||||
from utils.cache.files import read_text
|
||||
from utils.cache.yaml import load_yaml_any
|
||||
@@ -44,7 +44,7 @@ def jinja_env(searchpath) -> Environment:
|
||||
env = Environment(
|
||||
loader=FileSystemLoader(str(searchpath)),
|
||||
undefined=StrictUndefined,
|
||||
autoescape=False, # noqa: S701 - renders HCL and dotenv, not markup
|
||||
autoescape=select_autoescape(default_for_string=False),
|
||||
)
|
||||
env.filters["bool"] = bool
|
||||
env.filters["dotenv_quote"] = lambda value: f'"{value}"'
|
||||
|
||||
@@ -465,7 +465,7 @@ class TestResolveApiKey(unittest.TestCase):
|
||||
modules = {}
|
||||
backend.install(modules)
|
||||
with patch.dict("sys.modules", modules):
|
||||
return module, asyncio.run(module.resolve_api_key())
|
||||
return asyncio.run(module.resolve_api_key())
|
||||
|
||||
def test_an_existing_administrator_key_is_reused_and_kept(self) -> None:
|
||||
backend = FakeBackend(
|
||||
@@ -473,7 +473,7 @@ class TestResolveApiKey(unittest.TestCase):
|
||||
keys={"admin1": "sk-operator"},
|
||||
)
|
||||
|
||||
_module, (user_id, key, minted) = self._resolve(backend)
|
||||
user_id, key, minted = self._resolve(backend)
|
||||
|
||||
self.assertEqual("admin1", user_id)
|
||||
self.assertEqual("sk-operator", key)
|
||||
@@ -483,7 +483,7 @@ class TestResolveApiKey(unittest.TestCase):
|
||||
def test_a_missing_key_is_minted_and_marked_for_removal(self) -> None:
|
||||
backend = FakeBackend([FakeUser("admin1", "admin")])
|
||||
|
||||
_module, (user_id, key, minted) = self._resolve(backend)
|
||||
user_id, key, minted = self._resolve(backend)
|
||||
|
||||
self.assertEqual("admin1", user_id)
|
||||
self.assertTrue(key.startswith("sk-"))
|
||||
@@ -493,7 +493,7 @@ class TestResolveApiKey(unittest.TestCase):
|
||||
def test_an_instance_without_users_gets_one_administrator(self) -> None:
|
||||
backend = FakeBackend([])
|
||||
|
||||
_module, (user_id, _key, minted) = self._resolve(backend)
|
||||
user_id, _key, minted = self._resolve(backend)
|
||||
|
||||
self.assertEqual(["admin"], [user.role for user in backend.created])
|
||||
self.assertEqual(backend.created[0].id, user_id)
|
||||
@@ -512,9 +512,8 @@ class TestResolveApiKey(unittest.TestCase):
|
||||
admin_id, _key, minted = asyncio.run(module.resolve_api_key())
|
||||
self.assertTrue(minted)
|
||||
try:
|
||||
exploding_grant(_key)
|
||||
except RuntimeError:
|
||||
pass
|
||||
with self.assertRaises(RuntimeError):
|
||||
exploding_grant(_key)
|
||||
finally:
|
||||
asyncio.run(module.drop_api_key(admin_id))
|
||||
|
||||
|
||||
@@ -5,8 +5,8 @@ import struct
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
import unittest.mock as mock
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
from . import PROJECT_ROOT
|
||||
|
||||
|
||||
@@ -12,7 +12,7 @@ from __future__ import annotations
|
||||
import json
|
||||
import unittest
|
||||
|
||||
from jinja2 import Environment, StrictUndefined
|
||||
from jinja2 import Environment, StrictUndefined, select_autoescape
|
||||
|
||||
from utils.cache.files import read_text
|
||||
from utils.cache.yaml import load_yaml_any
|
||||
@@ -34,7 +34,7 @@ def _render(template_path, **variables) -> str:
|
||||
trim_blocks=True,
|
||||
lstrip_blocks=True,
|
||||
undefined=StrictUndefined,
|
||||
autoescape=False, # noqa: S701 - JMAP request bodies, never HTML
|
||||
autoescape=select_autoescape(default_for_string=False),
|
||||
)
|
||||
env.filters["to_json"] = json.dumps
|
||||
env.filters["int"] = int
|
||||
|
||||
@@ -16,7 +16,7 @@ from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
from jinja2 import Environment, StrictUndefined
|
||||
from jinja2 import Environment, StrictUndefined, select_autoescape
|
||||
|
||||
from utils.cache.files import read_text
|
||||
|
||||
@@ -41,7 +41,7 @@ def _render(base_url: str, *, sso_enabled: bool = True, socks_proxy: str = "") -
|
||||
trim_blocks=True,
|
||||
lstrip_blocks=True,
|
||||
undefined=StrictUndefined,
|
||||
autoescape=False, # noqa: S701 - a PHP config file, and Ansible's templar does not escape either
|
||||
autoescape=select_autoescape(default_for_string=False),
|
||||
)
|
||||
env.filters["bool"] = bool
|
||||
return env.from_string(read_text(TEMPLATE)).render(
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import unittest
|
||||
import unittest.mock as mock
|
||||
from typing import ClassVar
|
||||
from unittest import mock
|
||||
|
||||
from utils.cache.applications import get_variants
|
||||
from utils.roles.guide import guide_deployable, guide_variant, smallest_variant
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import contextlib
|
||||
import io
|
||||
import unittest
|
||||
from unittest import mock
|
||||
import unittest.mock as mock
|
||||
|
||||
from utils.tests.swarm import matrix
|
||||
|
||||
|
||||
Reference in new issue
Block a user