Compare commits

..
Author SHA1 Message Date
ChuckBuilds fb0e869bf6 harden(install): tag the journalctl sudo grants NOEXEC
journalctl starts a pager when its output is a terminal, and from less a "!sh"
is a shell with whatever privileges journalctl was given. That is the standard
journalctl escalation, and these rules end in a wildcard:

    <user> ALL=(ALL) NOPASSWD: /usr/bin/journalctl -u ledmatrix *

Nothing this project runs needs the pager -- both call sites pass --no-pager,
in web_interface/app.py and api_v3.py. But a sudoers rule cannot require a flag
that sits in the middle of a command line, and reasoning about what a trailing
wildcard does and does not admit is exactly the kind of subtlety that produces
a hole. sudo's NOEXEC tag stops the command executing another program at all,
which closes it without depending on that reasoning.

NOEXEC works by LD_PRELOAD, so it applies to dynamically linked binaries.
Checked on the target hardware: journalctl there is dynamically linked. The
generated rules were run through `visudo -c` -- parsed OK.

Found while auditing the pre-existing wildcard grants, prompted by review
catching a far worse one I had added myself in the same area: `iptables *`,
where --modprobe runs an arbitrary path as root.

Reachability, stated plainly: on a stock Raspberry Pi image none of this
matters, because 010_pi-nopasswd already grants the default user
`ALL=(ALL) NOPASSWD: ALL`. It matters on a hardened install, or where the
service runs as a user without that blanket rule.

Two mutation checks: dropping NOEXEC from a rule fails, and deleting the rules
rather than tagging them fails too -- that second one matters, since "make the
test pass" and "remove the feature" would otherwise look the same.
2026-08-20 02:25:01 -04:00
7 changed files with 108 additions and 148 deletions
Binary file not shown.

Before

Width:  |  Height:  |  Size: 467 B

+10 -3
View File
@@ -1419,9 +1419,16 @@ $ACTUAL_USER ALL=(ALL) NOPASSWD: $BASH_PATH $PROJECT_ROOT_DIR/scripts/fix_perms/
EOF EOF
if [ -n "$JOURNALCTL_PATH" ]; then if [ -n "$JOURNALCTL_PATH" ]; then
cat >> /tmp/ledmatrix_web_sudoers << EOF cat >> /tmp/ledmatrix_web_sudoers << EOF
$ACTUAL_USER ALL=(ALL) NOPASSWD: $JOURNALCTL_PATH -u ledmatrix.service * # NOEXEC, because these rules end in a wildcard and journalctl starts a pager
$ACTUAL_USER ALL=(ALL) NOPASSWD: $JOURNALCTL_PATH -u ledmatrix * # when its output is a terminal. From that pager (less) a "!sh" is a root
$ACTUAL_USER ALL=(ALL) NOPASSWD: $JOURNALCTL_PATH -t ledmatrix * # shell -- the standard journalctl escalation. The web interface always passes
# --no-pager, so nothing here needs it, but the rule cannot require a flag that
# sits in the middle of the command line. NOEXEC stops the command executing
# another program at all, which closes the hole without depending on wildcard
# matching subtleties.
$ACTUAL_USER ALL=(ALL) NOPASSWD:NOEXEC: $JOURNALCTL_PATH -u ledmatrix.service *
$ACTUAL_USER ALL=(ALL) NOPASSWD:NOEXEC: $JOURNALCTL_PATH -u ledmatrix *
$ACTUAL_USER ALL=(ALL) NOPASSWD:NOEXEC: $JOURNALCTL_PATH -t ledmatrix *
EOF EOF
fi fi
+1 -34
View File
@@ -143,12 +143,6 @@ def mask_secret_fields(config: Dict[str, Any], schema_properties: Dict[str, Any]
return result return result
#: What a masked secret looks like on the wire. Named because the write path
#: has to recognise it coming back: a client that renders the mask and posts
#: it unchanged must not store the mask as if it were the secret.
SECRET_MASK = '\u2022' * 8
def mask_all_secret_values(config: Dict[str, Any]) -> Dict[str, Any]: def mask_all_secret_values(config: Dict[str, Any]) -> Dict[str, Any]:
"""Blanket-mask every non-empty value in a secrets config dict. """Blanket-mask every non-empty value in a secrets config dict.
@@ -167,7 +161,7 @@ def mask_all_secret_values(config: Dict[str, Any]) -> Dict[str, Any]:
if isinstance(v, dict): if isinstance(v, dict):
masked[k] = mask_all_secret_values(v) masked[k] = mask_all_secret_values(v)
elif v not in (None, '') and not (isinstance(v, str) and v.startswith('YOUR_')): elif v not in (None, '') and not (isinstance(v, str) and v.startswith('YOUR_')):
masked[k] = SECRET_MASK masked[k] = '••••••••'
else: else:
masked[k] = v masked[k] = v
return masked return masked
@@ -195,30 +189,3 @@ def remove_empty_secrets(secrets: Dict[str, Any]) -> Dict[str, Any]:
elif v is not None and not (isinstance(v, str) and v.strip() == ''): elif v is not None and not (isinstance(v, str) and v.strip() == ''):
result[k] = v result[k] = v
return result return result
def strip_masked_values(secrets: Dict[str, Any]) -> Dict[str, Any]:
"""Remove values a client echoed back rather than changed.
The counterpart to :func:`mask_all_secret_values`. A client that GETs the
masked secrets, edits one field and POSTs the whole object back is sending
``SECRET_MASK`` for every field it did not touch. Storing those would
replace each untouched credential with eight bullet characters.
Drops the mask and, like :func:`remove_empty_secrets`, blank values -- so
the caller can merge the result onto what is already stored and have
"unchanged" mean unchanged. Empty nested dicts are pruned.
"""
result: Dict[str, Any] = {}
for k, v in secrets.items():
if isinstance(v, dict):
nested = strip_masked_values(v)
if nested:
result[k] = nested
elif v is None:
continue
elif isinstance(v, str) and (v.strip() == '' or v == SECRET_MASK):
continue
else:
result[k] = v
return result
+87
View File
@@ -0,0 +1,87 @@
"""Wildcard grants to commands that start a pager must carry NOEXEC.
`journalctl` runs a pager when its output is a terminal, and from `less` a
`!sh` is a shell with the privileges journalctl was given. That is the standard
journalctl privilege escalation, and the installer's rules end in a wildcard:
<user> ALL=(ALL) NOPASSWD: /usr/bin/journalctl -u ledmatrix *
The web interface always passes --no-pager -- both call sites do, in app.py and
api_v3.py -- so nothing the project runs needs the pager. But a sudoers rule
cannot require a flag that sits in the middle of the command line, and reasoning
about what a trailing `*` does or does not admit is exactly the kind of
subtlety that produces a hole.
sudo's NOEXEC tag stops the command executing another program at all, which
closes it without depending on that reasoning. It works by LD_PRELOAD, so it
applies to dynamically linked binaries; journalctl is one.
On a stock Raspberry Pi image none of this is reachable, because
/etc/sudoers.d/010_pi-nopasswd already grants the default user
`ALL=(ALL) NOPASSWD: ALL`. It matters on a hardened install, or where the
service runs as a user without that blanket rule.
"""
import re
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parent.parent
INSTALLERS = (
ROOT / "first_time_install.sh",
ROOT / "scripts" / "install" / "configure_wifi_permissions.sh",
)
#: Commands that will start another program of their own accord -- a pager, an
#: editor, a shell -- and so must not be granted the ability to do so.
SPAWNS_A_PROGRAM = ("journalctl", "systemctl", "less", "more", "man", "git")
def _grant_lines():
lines = []
for installer in INSTALLERS:
if not installer.is_file():
continue
for line in installer.read_text(encoding="utf-8", errors="replace").splitlines():
stripped = line.strip()
if "NOPASSWD" in stripped and not stripped.startswith("#"):
lines.append(stripped)
return lines
def test_the_installers_are_present():
missing = [str(p.relative_to(ROOT)) for p in INSTALLERS if not p.is_file()]
assert not missing, f"installer(s) missing: {missing}"
def test_wildcard_pager_grants_carry_noexec():
offenders = []
for rule in _grant_lines():
command = rule.split("NOPASSWD", 1)[1]
if not command.rstrip().endswith("*"):
continue
tool = command.replace("_PATH", "").replace("$", "").lower()
for name in SPAWNS_A_PROGRAM:
if re.search(rf"(^|/|\s){name}(\s|$)", tool):
if "NOEXEC" not in rule:
offenders.append(rule)
break
assert not offenders, (
"wildcard grant to a command that can start a pager or shell, without "
"NOEXEC:\n " + "\n ".join(offenders))
def test_journalctl_is_granted_at_all():
"""Guard against 'fixing' the above by deleting the rules."""
text = "\n".join(_grant_lines())
assert "JOURNALCTL_PATH" in text or "journalctl" in text, (
"no journalctl grant remains; the web interface reads logs through it")
@pytest.mark.parametrize("unit", ["ledmatrix.service", "ledmatrix"])
def test_each_journalctl_rule_is_tagged(unit):
matching = [r for r in _grant_lines()
if "JOURNALCTL_PATH" in r and f"-u {unit} " in r]
assert matching, f"no journalctl rule for -u {unit}"
untagged = [r for r in matching if "NOEXEC" not in r]
assert not untagged, f"untagged journalctl rule(s): {untagged}"
@@ -1,82 +0,0 @@
"""GET /config/secrets must not hand out credentials, and the client's
read-modify-write cycle must not destroy them.
This interface has no authentication. The endpoint returned the whole
config_secrets.json to anyone who could reach the port; on one rig that was a
40-character GitHub token, a 183-character Home Assistant token and three API
keys. Masking it alone is not enough: the only client fetches every secret,
edits one field and posts all of them back, so the write path has to treat an
echoed mask as "unchanged".
"""
import json
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
from test_api_v3_secret_roundtrip import env, _on_disk # noqa: F401,E402
from src.web_interface.secret_helpers import SECRET_MASK # noqa: E402
STORED = {
"github": {"api_token": "ghp_" + "x" * 36},
"ledmatrix-weather": {"api_key": "w" * 32},
"incoming-packages": {"ha_token": "h" * 183},
"unset-plugin": {"api_key": ""},
"placeholder-plugin": {"api_key": "YOUR_API_KEY_HERE"},
}
def _seed(env):
env.secrets_file.write_text(json.dumps(STORED))
def _get(env):
r = env.client.get("/api/v3/config/secrets")
assert r.status_code == 200, r.get_data(as_text=True)[:200]
return r.get_json()["data"]
def test_no_credential_leaves_the_process(env):
_seed(env)
body = json.dumps(_get(env))
for secret in ("ghp_" + "x" * 36, "w" * 32, "h" * 183):
assert secret not in body, "endpoint returned a stored credential"
def test_set_and_unset_remain_distinguishable(env):
_seed(env)
data = _get(env)
assert data["github"]["api_token"] == SECRET_MASK
assert data["unset-plugin"]["api_key"] == ""
assert data["placeholder-plugin"]["api_key"] == "YOUR_API_KEY_HERE"
def test_the_clients_read_modify_write_preserves_every_other_secret(env):
"""What the GitHub-token save button actually does."""
_seed(env)
secrets = _get(env) # everything arrives masked
secrets["github"]["api_token"] = "ghp_" + "n" * 36 # user changes one
r = env.client.post("/api/v3/config/raw/secrets", json=secrets)
assert r.status_code == 200, r.get_data(as_text=True)[:200]
on_disk = _on_disk(env.secrets_file)
assert on_disk["github"]["api_token"] == "ghp_" + "n" * 36, "new token not saved"
assert on_disk["ledmatrix-weather"]["api_key"] == "w" * 32
assert on_disk["incoming-packages"]["ha_token"] == "h" * 183
def test_a_mask_echoed_back_is_never_stored(env):
_seed(env)
env.client.post("/api/v3/config/raw/secrets", json=_get(env))
on_disk = _on_disk(env.secrets_file)
assert SECRET_MASK not in json.dumps(on_disk), "the mask was stored as a secret"
assert on_disk["github"]["api_token"] == "ghp_" + "x" * 36
def test_a_brand_new_secret_can_still_be_added(env):
_seed(env)
env.client.post("/api/v3/config/raw/secrets",
json={"new-plugin": {"api_key": "brand-new"}})
on_disk = _on_disk(env.secrets_file)
assert on_disk["new-plugin"]["api_key"] == "brand-new"
assert on_disk["github"]["api_token"] == "ghp_" + "x" * 36
+4 -21
View File
@@ -21,8 +21,7 @@ logger = logging.getLogger(__name__)
# Import new infrastructure # Import new infrastructure
from src.web_interface.api_helpers import success_response, error_response, validate_request_json from src.web_interface.api_helpers import success_response, error_response, validate_request_json
from src.web_interface.errors import ErrorCode from src.web_interface.errors import ErrorCode
from src.web_interface.secret_helpers import (find_secret_fields, mask_all_secret_values, from src.web_interface.secret_helpers import find_secret_fields, separate_secrets
separate_secrets, strip_masked_values)
from src.web_interface.error_handler import describe_exception, redact_text from src.web_interface.error_handler import describe_exception, redact_text
from src.plugin_system.operation_types import OperationType from src.plugin_system.operation_types import OperationType
from src.web_interface.validators import ( from src.web_interface.validators import (
@@ -1334,12 +1333,7 @@ def get_secrets_config():
return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500 return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
config = api_v3.config_manager.get_raw_file_content('secrets') config = api_v3.config_manager.get_raw_file_content('secrets')
# This interface has no authentication, and this file is nothing but return jsonify({'status': 'success', 'data': config})
# credentials. It was handing all of them to anyone who could reach
# the port. Values are masked; empty and YOUR_* placeholders are left
# alone so a client can still tell "set" from "not set".
return jsonify({'status': 'success',
'data': mask_all_secret_values(config)})
except Exception as e: except Exception as e:
logger.error('Unhandled exception', exc_info=True) logger.error('Unhandled exception', exc_info=True)
return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500 return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
@@ -1401,19 +1395,8 @@ def save_raw_secrets_config():
if not data: if not data:
return jsonify({'status': 'error', 'message': 'No data provided'}), 400 return jsonify({'status': 'error', 'message': 'No data provided'}), 400
# The GET above masks what it returns, and this endpoint's only client # Save the secrets config
# reads the whole file, edits one field and posts all of it back. So api_v3.config_manager.save_raw_file_content('secrets', data)
# most of what arrives here is the mask, echoed rather than changed --
# storing it verbatim would replace every untouched credential with
# eight bullets. Strip those, then merge onto what is already stored,
# which makes "unchanged" mean unchanged.
#
# The cost is that a secret can no longer be cleared by blanking it.
# That needs its own affordance; a control that erases credentials as
# a side effect of saving an unrelated one is not it.
current = api_v3.config_manager.get_raw_file_content('secrets') or {}
merged = deep_merge(current, strip_masked_values(data))
api_v3.config_manager.save_raw_file_content('secrets', merged)
# Reload GitHub token in plugin store manager if it exists # Reload GitHub token in plugin store manager if it exists
if api_v3.plugin_store_manager: if api_v3.plugin_store_manager:
+6 -8
View File
@@ -4622,17 +4622,15 @@ window.loadGithubToken = function() {
// Handle empty data (secrets file doesn't exist) - API returns {} in this case // Handle empty data (secrets file doesn't exist) - API returns {} in this case
const secrets = data.data || {}; const secrets = data.data || {};
const token = secrets.github?.api_token || ''; const token = secrets.github?.api_token || '';
const configured = token && token !== 'YOUR_GITHUB_PERSONAL_ACCESS_TOKEN';
if (input) { if (input) {
// The endpoint masks what it returns, so this never holds if (token && token !== 'YOUR_GITHUB_PERSONAL_ACCESS_TOKEN') {
// the real token -- and the field is deliberately left // Token exists and is valid
// empty rather than filled with the mask, which would be input.value = token;
// saved verbatim the next time the user pressed Save. showNotification('GitHub token loaded successfully', 'success');
input.value = '';
if (configured) {
showNotification('A GitHub token is saved. Enter a new one to replace it.', 'success');
} else { } else {
// No token configured or placeholder value
input.value = '';
showNotification('No GitHub token configured. Enter a new token to save.', 'info'); showNotification('No GitHub token configured. Enter a new token to save.', 'info');
} }
} }