mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-04 06:15:09 +00:00
* fix(core): font zip cache, monotonic timers, resolver back-off, and other core/common fixes - font_manager: a .zip font URL is served as its extracted font after a restart (the cached-file check returned the archive first); downloads use requests with a 30s timeout into a temp file + os.replace. - api_helper / sync_manager: rate-limit and heartbeat/leader timeouts use time.monotonic(); last_request_time and the status file's ts stay wall-clock. set_on_new_cycle docstring no longer claims core uses it. - logo_helper: the placeholder uses the same scaled box as a real logo. - permission_utils: one _sudo_bash_candidates() helper (with the sudoers exact-argv rationale) shared by sudo_remove_directory, which now retries the next bash path on a sudo refusal, and install_requirements_file. - dynamic_team_resolver: failed/empty fetch backs off 5 min; duplicate INFO log and contradictory docstring example fixed. - element_style: scale default looked up through element aliases. - background_data_service: cache-hit callback runs outside the lock. - config_arrays: union-aware type check (["array","null"]); stale dotToNested() reference removed. - auto_update_setup: non-dict auto_update reads as off; temp result file unlinked when the write fails. - exceptions: constructors copy the caller's context dict. - logging_config: StructuredFormatter json.dumps(default=str). - error_aggregator: removed unused export_path/export_to_file/_auto_export. - Docstrings: validate_file_upload max_size_mb, raise_on_errors. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(sync): retry the status-file rename like the other atomic writers On Windows os.replace can fail with "Access is denied" while a scanner briefly holds the target open; config_manager_atomic._replace already retries that (and re-raises at once on other platforms). The sync status writer called os.replace directly, which made test_concurrent_writers_each_use_their_own_temp_file flaky on Windows. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
174 lines
7.2 KiB
Python
174 lines
7.2 KiB
Python
"""
|
|
Tests for src.common.permission_utils.
|
|
|
|
Covers two things:
|
|
|
|
* URL-credential redaction -- the fix for a CodeQL clear-text-logging-of-secrets
|
|
alert: install_requirements_file() must never let a private index URL's
|
|
embedded user:pass@ credentials reach logs or its returned CompletedProcess,
|
|
since pip can echo that URL back verbatim in its own stderr/stdout on failure.
|
|
* ensure_shared_group_ownership() staying a silent no-op on platforms without
|
|
the POSIX ownership APIs.
|
|
"""
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from src.common.permission_utils import (
|
|
_redact_url_credentials,
|
|
ensure_shared_group_ownership,
|
|
install_requirements_file,
|
|
sudo_remove_directory,
|
|
)
|
|
|
|
|
|
class TestRedactUrlCredentials:
|
|
def test_redacts_embedded_basic_auth(self):
|
|
text = "Could not fetch URL https://alice:s3cr3t@pypi.example.com/simple/: 403"
|
|
redacted = _redact_url_credentials(text)
|
|
assert "s3cr3t" not in redacted
|
|
assert "alice" not in redacted
|
|
assert "https://***:***@pypi.example.com/simple/" in redacted
|
|
|
|
def test_leaves_credential_free_text_unchanged(self):
|
|
text = "ERROR: Could not find a version that satisfies the requirement foo==1.0"
|
|
assert _redact_url_credentials(text) == text
|
|
|
|
def test_handles_none_and_empty(self):
|
|
assert _redact_url_credentials(None) == ""
|
|
assert _redact_url_credentials("") == ""
|
|
|
|
def test_does_not_touch_denied_check_phrases(self):
|
|
"""The fixed phrases install_requirements_file greps for must survive
|
|
redaction untouched -- they don't overlap with URL syntax, but this
|
|
pins that assumption so a regex change can't silently break it."""
|
|
text = "sudo: a password is required"
|
|
assert _redact_url_credentials(text) == text
|
|
|
|
|
|
class TestSudoRemoveDirectory:
|
|
"""sudoers matches the exact argv, so the bash path the rule names has
|
|
to be found by trying each candidate, as install_requirements_file does."""
|
|
|
|
def _target(self, tmp_path):
|
|
target = tmp_path / "some-plugin"
|
|
target.mkdir()
|
|
return target
|
|
|
|
@patch('src.common.permission_utils.subprocess.run')
|
|
def test_tries_the_next_bash_path_when_sudo_refuses(self, mock_run, tmp_path):
|
|
target = self._target(tmp_path)
|
|
|
|
def fake_run(argv, **kwargs):
|
|
if mock_run.call_count == 1:
|
|
return MagicMock(returncode=1, stdout="",
|
|
stderr="sudo: a password is required")
|
|
target.rmdir()
|
|
return MagicMock(returncode=0, stdout="", stderr="")
|
|
mock_run.side_effect = fake_run
|
|
|
|
assert sudo_remove_directory(target, allowed_bases=[tmp_path]) is True
|
|
assert mock_run.call_count == 2
|
|
first, second = (c.args[0][2] for c in mock_run.call_args_list)
|
|
assert first != second
|
|
|
|
@patch('src.common.permission_utils.subprocess.run')
|
|
def test_stops_when_the_helper_itself_fails(self, mock_run, tmp_path):
|
|
target = self._target(tmp_path)
|
|
mock_run.return_value = MagicMock(returncode=1, stdout="",
|
|
stderr="refusing: not a plugin dir")
|
|
|
|
assert sudo_remove_directory(target, allowed_bases=[tmp_path]) is False
|
|
assert mock_run.call_count == 1
|
|
|
|
|
|
class TestInstallRequirementsFileRedaction:
|
|
@patch('src.common.permission_utils.subprocess.run')
|
|
def test_wrapper_path_redacts_stderr_and_stdout(self, mock_run, tmp_path):
|
|
"""safe_pip_install.sh exists in this repo, so install_requirements_file
|
|
takes the sudo-wrapper branch; a failing result must come back
|
|
with any embedded index-URL credentials already redacted."""
|
|
req_file = tmp_path / "requirements.txt"
|
|
req_file.write_text("requests\n")
|
|
|
|
mock_run.return_value = MagicMock(
|
|
returncode=1,
|
|
stdout="Looking in indexes: https://bob:hunter2@pypi.internal/simple\n",
|
|
stderr="ERROR https://bob:hunter2@pypi.internal/simple/foo: 401",
|
|
)
|
|
|
|
result = install_requirements_file(req_file, timeout=5)
|
|
|
|
assert "hunter2" not in result.stdout
|
|
assert "hunter2" not in result.stderr
|
|
assert "https://***:***@pypi.internal" in result.stdout
|
|
assert "https://***:***@pypi.internal" in result.stderr
|
|
|
|
@patch('src.common.permission_utils.subprocess.run')
|
|
@patch('src.common.permission_utils.Path.exists', return_value=False)
|
|
def test_no_wrapper_fallback_path_redacts_stderr_and_stdout(self, mock_exists, mock_run, tmp_path):
|
|
"""No safe_pip_install.sh wrapper -> falls straight to the
|
|
sys.executable pip fallback (the second subprocess.run call site);
|
|
its result must come back redacted too, independent of the wrapper
|
|
branch's own redaction above."""
|
|
req_file = tmp_path / "requirements.txt"
|
|
req_file.write_text("requests\n")
|
|
|
|
mock_run.return_value = MagicMock(
|
|
returncode=1,
|
|
stdout="Looking in indexes: https://carol:swordfish@pypi.internal/simple\n",
|
|
stderr="ERROR https://carol:swordfish@pypi.internal/simple/foo: 401",
|
|
)
|
|
|
|
result = install_requirements_file(req_file, timeout=5)
|
|
|
|
assert "swordfish" not in result.stdout
|
|
assert "swordfish" not in result.stderr
|
|
assert "https://***:***@pypi.internal" in result.stdout
|
|
assert "https://***:***@pypi.internal" in result.stderr
|
|
|
|
|
|
class TestEnsureSharedGroupOwnership:
|
|
"""The chgrp self-heal must stay a no-op wherever it cannot apply.
|
|
|
|
``ConfigManager.load_config()`` calls this on every load that finds a
|
|
secrets file, and its callers only ever catch ``OSError``. An
|
|
``AttributeError`` from looking up a POSIX-only name on Windows therefore
|
|
escaped all the way out of ``load_config``, and took the import of
|
|
``web_interface.app`` with it on any Windows checkout that had a
|
|
``config/config_secrets.json``.
|
|
"""
|
|
|
|
def test_no_geteuid_is_a_silent_no_op(self, monkeypatch, tmp_path):
|
|
secrets = tmp_path / "config_secrets.json"
|
|
secrets.write_text("{}", encoding='utf-8')
|
|
monkeypatch.delattr(os, "geteuid", raising=False)
|
|
monkeypatch.delattr(os, "chown", raising=False)
|
|
|
|
ensure_shared_group_ownership(secrets) # must not raise
|
|
|
|
def test_non_root_never_chowns(self, monkeypatch, tmp_path):
|
|
chown = MagicMock()
|
|
monkeypatch.setattr(os, "geteuid", lambda: 1000, raising=False)
|
|
monkeypatch.setattr(os, "chown", chown, raising=False)
|
|
|
|
ensure_shared_group_ownership(tmp_path / "config_secrets.json")
|
|
|
|
chown.assert_not_called()
|
|
|
|
def test_root_chowns_a_file_whose_group_is_wrong(self, monkeypatch, tmp_path):
|
|
secrets = tmp_path / "config_secrets.json"
|
|
secrets.write_text("{}", encoding='utf-8')
|
|
# Any gid the file does not already have, so the chgrp is due.
|
|
wanted = secrets.stat().st_gid + 1
|
|
chown = MagicMock()
|
|
monkeypatch.setattr(os, "geteuid", lambda: 0, raising=False)
|
|
monkeypatch.setattr(os, "chown", chown, raising=False)
|
|
monkeypatch.setattr('src.common.permission_utils.get_shared_group_gid',
|
|
lambda: wanted)
|
|
|
|
ensure_shared_group_ownership(secrets)
|
|
|
|
chown.assert_called_once_with(secrets, -1, wanted)
|