mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-12 01:56:34 +00:00
Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e395ebe008 | ||
|
|
c73d7a0399 |
@@ -125,19 +125,6 @@ New modules a plugin may import via `src.*` (floor on 3.5.0):
|
|||||||
fetched, which keeps the peak memory of a four-capped-month fetch to about
|
fetched, which keeps the peak memory of a four-capped-month fetch to about
|
||||||
16 MB over the sequential path rather than 43 MB — `docs/LOW_MEMORY_BOARDS.md`
|
16 MB over the sequential path rather than 43 MB — `docs/LOW_MEMORY_BOARDS.md`
|
||||||
puts a 1 GB Pi 3B+ at under 200 MB of headroom.
|
puts a 1 GB Pi 3B+ at under 200 MB of headroom.
|
||||||
- `ESPNDataSource.fetch_standings` asks each league the endpoint that league
|
|
||||||
actually publishes. It tried `/standings` first whatever the league and fell
|
|
||||||
back to `/rankings` only on a 404, but college leagues answer `/standings`
|
|
||||||
with a 200 that carries no poll, so the fallback never fired: the rank badge
|
|
||||||
simply never appeared and anything keyed off rankings quietly did nothing.
|
|
||||||
Endpoints are now ordered by whether the league publishes a poll, and a 200
|
|
||||||
that lacks the key counts as a miss, so a league answering both still ends up
|
|
||||||
with whichever carries the poll. Only a 404 is routine — that is how a league
|
|
||||||
says it has none; a connection error, a timeout or an unparseable body is
|
|
||||||
logged as an error again, and a bug raised while inspecting the payload is no
|
|
||||||
longer swallowed as a missing poll. This is the implementation the football,
|
|
||||||
baseball and hockey boards already ship; core was the last copy on the old
|
|
||||||
one.
|
|
||||||
|
|
||||||
### Scrolling
|
### Scrolling
|
||||||
|
|
||||||
@@ -367,19 +354,6 @@ New modules a plugin may import via `src.*` (floor on 3.5.0):
|
|||||||
budget, so a rollback finishes inside the unit's 30-minute limit instead of
|
budget, so a rollback finishes inside the unit's 30-minute limit instead of
|
||||||
being killed mid-way.
|
being killed mid-way.
|
||||||
|
|
||||||
### Installers
|
|
||||||
|
|
||||||
- The generated `ledmatrix_web` sudoers rules are parsed before they are
|
|
||||||
installed. Both installers built the drop-in from `which` lookups and copied
|
|
||||||
it into `/etc/sudoers.d` without ever checking it, and a malformed file there
|
|
||||||
makes sudo refuse every command for every user — on a headless Pi, that is
|
|
||||||
unrecoverable over SSH. `first_time_install.sh` now runs `visudo -c` on the
|
|
||||||
generated file and, if it does not parse, prints what visudo said and leaves
|
|
||||||
the installed file untouched instead of replacing it with a broken one;
|
|
||||||
`configure_web_sudo.sh` does the same before offering the rules for
|
|
||||||
confirmation. `first_time_install.sh` also built that file at a fixed `/tmp`
|
|
||||||
path as root; `mktemp` now picks the name.
|
|
||||||
|
|
||||||
### Small fixes (update-all, plugin system settings, scripts)
|
### Small fixes (update-all, plugin system settings, scripts)
|
||||||
|
|
||||||
- **Check & Update All** counts a plugin that had nothing to update as
|
- **Check & Update All** counts a plugin that had nothing to update as
|
||||||
|
|||||||
+7
-31
@@ -1504,9 +1504,6 @@ echo "------------------------------------------------"
|
|||||||
# Create sudoers configuration for the web interface
|
# Create sudoers configuration for the web interface
|
||||||
echo "Creating sudoers configuration..."
|
echo "Creating sudoers configuration..."
|
||||||
SUDOERS_FILE="/etc/sudoers.d/ledmatrix_web"
|
SUDOERS_FILE="/etc/sudoers.d/ledmatrix_web"
|
||||||
# A predictable name in a world-writable directory is a symlink target;
|
|
||||||
# root writes the rules here, so let mktemp pick the name.
|
|
||||||
SUDOERS_TMP=$(mktemp "${TMPDIR:-/tmp}/ledmatrix_web_sudoers.XXXXXX")
|
|
||||||
|
|
||||||
# Get command paths
|
# Get command paths
|
||||||
PYTHON_PATH=$(which python3)
|
PYTHON_PATH=$(which python3)
|
||||||
@@ -1517,7 +1514,7 @@ BASH_PATH=$(which bash)
|
|||||||
JOURNALCTL_PATH=$(which journalctl 2>/dev/null || true)
|
JOURNALCTL_PATH=$(which journalctl 2>/dev/null || true)
|
||||||
|
|
||||||
# Create sudoers content
|
# Create sudoers content
|
||||||
cat > "$SUDOERS_TMP" << EOF
|
cat > /tmp/ledmatrix_web_sudoers << EOF
|
||||||
# LED Matrix Web Interface passwordless sudo configuration
|
# LED Matrix Web Interface passwordless sudo configuration
|
||||||
# This allows the web interface user to run specific commands without a password
|
# This allows the web interface user to run specific commands without a password
|
||||||
|
|
||||||
@@ -1544,7 +1541,7 @@ $ACTUAL_USER ALL=(ALL) NOPASSWD: $BASH_PATH $PROJECT_ROOT_DIR/scripts/fix_perms/
|
|||||||
$ACTUAL_USER ALL=(ALL) NOPASSWD: $BASH_PATH $PROJECT_ROOT_DIR/scripts/fix_perms/safe_pip_install.sh *
|
$ACTUAL_USER ALL=(ALL) NOPASSWD: $BASH_PATH $PROJECT_ROOT_DIR/scripts/fix_perms/safe_pip_install.sh *
|
||||||
EOF
|
EOF
|
||||||
if [ -n "$JOURNALCTL_PATH" ]; then
|
if [ -n "$JOURNALCTL_PATH" ]; then
|
||||||
cat >> "$SUDOERS_TMP" << EOF
|
cat >> /tmp/ledmatrix_web_sudoers << EOF
|
||||||
# NOEXEC, because these rules end in a wildcard and journalctl starts a pager
|
# NOEXEC, because these rules end in a wildcard and journalctl starts a pager
|
||||||
# when its output is a terminal. From that pager (less) a "!sh" is a root
|
# when its output is a terminal. From that pager (less) a "!sh" is a root
|
||||||
# shell -- the standard journalctl escalation. The web interface always passes
|
# shell -- the standard journalctl escalation. The web interface always passes
|
||||||
@@ -1558,38 +1555,17 @@ $ACTUAL_USER ALL=(ALL) NOPASSWD:NOEXEC: $JOURNALCTL_PATH -t ledmatrix *
|
|||||||
EOF
|
EOF
|
||||||
fi
|
fi
|
||||||
|
|
||||||
# Never install rules we have not parsed. A malformed drop-in in
|
if [ -f "$SUDOERS_FILE" ] && cmp -s /tmp/ledmatrix_web_sudoers "$SUDOERS_FILE"; then
|
||||||
# /etc/sudoers.d makes sudo refuse every command for every user, which on a
|
|
||||||
# headless Pi leaves no way in at all. If the rules do not parse, say so and
|
|
||||||
# keep whatever is already installed.
|
|
||||||
SUDOERS_VALID=1
|
|
||||||
if command -v visudo >/dev/null 2>&1; then
|
|
||||||
if ! visudo -c -f "$SUDOERS_TMP" >/dev/null 2>&1; then
|
|
||||||
SUDOERS_VALID=0
|
|
||||||
echo "⚠ The generated sudoers rules did not parse:" >&2
|
|
||||||
visudo -c -f "$SUDOERS_TMP" >&2 || true
|
|
||||||
echo "⚠ Leaving $SUDOERS_FILE unchanged. The web interface cannot control" >&2
|
|
||||||
echo " the display service until this is fixed." >&2
|
|
||||||
fi
|
|
||||||
else
|
|
||||||
echo "⚠ visudo not found; installing the sudoers rules unvalidated"
|
|
||||||
fi
|
|
||||||
|
|
||||||
if [ "$SUDOERS_VALID" = "0" ]; then
|
|
||||||
rm -f "$SUDOERS_TMP"
|
|
||||||
elif [ -f "$SUDOERS_FILE" ] && cmp -s "$SUDOERS_TMP" "$SUDOERS_FILE"; then
|
|
||||||
echo "Sudoers configuration already up to date"
|
echo "Sudoers configuration already up to date"
|
||||||
rm -f "$SUDOERS_TMP"
|
rm /tmp/ledmatrix_web_sudoers
|
||||||
else
|
else
|
||||||
echo "Installing/updating sudoers configuration..."
|
echo "Installing/updating sudoers configuration..."
|
||||||
cp "$SUDOERS_TMP" "$SUDOERS_FILE"
|
cp /tmp/ledmatrix_web_sudoers "$SUDOERS_FILE"
|
||||||
chmod 440 "$SUDOERS_FILE"
|
chmod 440 "$SUDOERS_FILE"
|
||||||
rm -f "$SUDOERS_TMP"
|
rm /tmp/ledmatrix_web_sudoers
|
||||||
fi
|
fi
|
||||||
|
|
||||||
if [ "$SUDOERS_VALID" = "1" ]; then
|
echo "✓ Passwordless sudo access configured"
|
||||||
echo "✓ Passwordless sudo access configured"
|
|
||||||
fi
|
|
||||||
echo ""
|
echo ""
|
||||||
|
|
||||||
CURRENT_STEP="Configure WiFi management permissions"
|
CURRENT_STEP="Configure WiFi management permissions"
|
||||||
|
|||||||
@@ -130,19 +130,6 @@ TEMP_SUDOERS="/tmp/ledmatrix_web_sudoers_$$"
|
|||||||
echo "$WEB_USER ALL=(ALL) NOPASSWD: $BASH_PATH $SAFE_PIP_INSTALL_PATH *"
|
echo "$WEB_USER ALL=(ALL) NOPASSWD: $BASH_PATH $SAFE_PIP_INSTALL_PATH *"
|
||||||
} > "$TEMP_SUDOERS"
|
} > "$TEMP_SUDOERS"
|
||||||
|
|
||||||
# Never offer to install rules we have not parsed. A malformed drop-in in
|
|
||||||
# /etc/sudoers.d makes sudo refuse every command for every user.
|
|
||||||
if command -v visudo >/dev/null 2>&1; then
|
|
||||||
if ! visudo -c -f "$TEMP_SUDOERS" >/dev/null 2>&1; then
|
|
||||||
echo ""
|
|
||||||
echo "✗ The generated sudoers rules did not parse:" >&2
|
|
||||||
visudo -c -f "$TEMP_SUDOERS" >&2 || true
|
|
||||||
echo "Nothing was changed." >&2
|
|
||||||
rm -f "$TEMP_SUDOERS"
|
|
||||||
exit 1
|
|
||||||
fi
|
|
||||||
fi
|
|
||||||
|
|
||||||
echo ""
|
echo ""
|
||||||
echo "Generated sudoers configuration:"
|
echo "Generated sudoers configuration:"
|
||||||
echo "--------------------------------"
|
echo "--------------------------------"
|
||||||
|
|||||||
@@ -114,68 +114,35 @@ class ESPNDataSource(DataSource):
|
|||||||
return []
|
return []
|
||||||
|
|
||||||
def fetch_standings(self, sport: str, league: str) -> Dict:
|
def fetch_standings(self, sport: str, league: str) -> Dict:
|
||||||
"""Fetch standings, or the poll for leagues that have one.
|
"""Fetch standings from ESPN API."""
|
||||||
|
# Try standings endpoint first (for professional leagues like NFL, NBA, etc.)
|
||||||
Order matters and used to be wrong. College leagues publish a poll at
|
try:
|
||||||
/rankings and a records table at /standings; professional leagues have
|
url = f"{self.base_url}/{sport}/{league}/standings"
|
||||||
only /standings. The old code tried /standings first and fell back to
|
response = self.session.get(url, headers=self.get_headers(), timeout=15)
|
||||||
/rankings only on a 404 -- but college /standings answers 200, so the
|
response.raise_for_status()
|
||||||
fallback never fired and college rankings came back empty forever.
|
|
||||||
Nothing failed; the AP rank badge simply never appeared, and anything
|
data = response.json()
|
||||||
else keyed off rankings quietly did nothing.
|
self.logger.debug(f"Fetched standings for {sport}/{league}")
|
||||||
|
|
||||||
A 200 that lacks the key is treated as a miss, so a league answering
|
|
||||||
both endpoints still ends up with whichever one actually carries a poll.
|
|
||||||
"""
|
|
||||||
league_name = (league or "").lower()
|
|
||||||
wants_poll = "college" in league_name or "ncaa" in league_name
|
|
||||||
endpoints = ["rankings", "standings"] if wants_poll else ["standings", "rankings"]
|
|
||||||
|
|
||||||
for endpoint in endpoints:
|
|
||||||
url = f"{self.base_url}/{sport}/{league}/{endpoint}"
|
|
||||||
# Only the request is guarded. Inspecting the payload happens
|
|
||||||
# below, outside the handler, so that a bug in this method cannot
|
|
||||||
# be mistaken for an endpoint that failed -- that mistake would
|
|
||||||
# silently drop rankings for a league that has them, which is the
|
|
||||||
# exact failure this function was written to fix.
|
|
||||||
try:
|
|
||||||
response = self.session.get(
|
|
||||||
url, headers=self.get_headers(), timeout=15
|
|
||||||
)
|
|
||||||
response.raise_for_status()
|
|
||||||
data = response.json()
|
|
||||||
except (requests.RequestException, ValueError) as e:
|
|
||||||
status = getattr(getattr(e, "response", None), "status_code", None)
|
|
||||||
# Only a 404 is routine -- it is how a league says "no poll
|
|
||||||
# here". Everything else is worth an error, and `status is
|
|
||||||
# None` covers the ones that matter most: ConnectionError,
|
|
||||||
# Timeout, a body that would not parse. Silencing those left a
|
|
||||||
# board that could not reach ESPN with one debug line, and the
|
|
||||||
# ranked filter running on an empty table.
|
|
||||||
if status != 404:
|
|
||||||
self.logger.error(
|
|
||||||
f"Error fetching {endpoint} from ESPN for "
|
|
||||||
f"{sport}/{league}: {e}"
|
|
||||||
)
|
|
||||||
continue
|
|
||||||
|
|
||||||
if not isinstance(data, dict):
|
|
||||||
# A list or a bare string is not something the callers can
|
|
||||||
# read. Treat it as a miss so the other endpoint still gets a
|
|
||||||
# turn, but say so -- this means ESPN changed shape.
|
|
||||||
self.logger.error(
|
|
||||||
f"Unexpected {endpoint} payload for {sport}/{league}: "
|
|
||||||
f"got {type(data).__name__}, expected an object"
|
|
||||||
)
|
|
||||||
continue
|
|
||||||
if endpoint == "rankings" and not data.get("rankings"):
|
|
||||||
continue
|
|
||||||
self.logger.debug(f"Fetched {endpoint} for {sport}/{league}")
|
|
||||||
return data
|
return data
|
||||||
self.logger.debug(
|
except Exception as e:
|
||||||
f"Standings/rankings not available for {sport}/{league} from ESPN API"
|
# If standings doesn't exist, try rankings (for college sports)
|
||||||
)
|
if hasattr(e, 'response') and hasattr(e.response, 'status_code') and e.response.status_code == 404:
|
||||||
return {}
|
try:
|
||||||
|
url = f"{self.base_url}/{sport}/{league}/rankings"
|
||||||
|
response = self.session.get(url, headers=self.get_headers(), timeout=15)
|
||||||
|
response.raise_for_status()
|
||||||
|
|
||||||
|
data = response.json()
|
||||||
|
self.logger.debug(f"Fetched rankings for {sport}/{league}")
|
||||||
|
return data
|
||||||
|
except Exception:
|
||||||
|
# Both endpoints failed - standings/rankings may not be available for this sport/league
|
||||||
|
self.logger.debug(f"Standings/rankings not available for {sport}/{league} from ESPN API")
|
||||||
|
return {}
|
||||||
|
else:
|
||||||
|
# Non-404 error - log at debug level since standings are optional
|
||||||
|
self.logger.debug(f"Error fetching standings from ESPN for {sport}/{league}: {e}")
|
||||||
|
return {}
|
||||||
|
|
||||||
|
|
||||||
class MLBAPIDataSource(DataSource):
|
class MLBAPIDataSource(DataSource):
|
||||||
|
|||||||
@@ -92,108 +92,10 @@ class TestESPNDataSource:
|
|||||||
assert result == payload
|
assert result == payload
|
||||||
|
|
||||||
def test_fetch_standings_returns_empty_on_error(self):
|
def test_fetch_standings_returns_empty_on_error(self):
|
||||||
# A transport failure is a RequestException, not a bare Exception.
|
with patch.object(self.source.session, "get", side_effect=Exception("error")):
|
||||||
# The old stand-in passed only because the handler caught everything,
|
|
||||||
# including bugs in the method under test.
|
|
||||||
with patch.object(self.source.session, "get",
|
|
||||||
side_effect=requests.ConnectionError("error")):
|
|
||||||
result = self.source.fetch_standings("football", "nfl")
|
result = self.source.fetch_standings("football", "nfl")
|
||||||
assert result == {}
|
assert result == {}
|
||||||
|
|
||||||
# ------------------------------------------------------------------
|
|
||||||
# fetch_standings endpoint selection
|
|
||||||
#
|
|
||||||
# College leagues publish a poll at /rankings and a records table at
|
|
||||||
# /standings; professional leagues have only /standings. Probing them in
|
|
||||||
# the wrong order still returns 200 -- just without a poll in it -- so
|
|
||||||
# nothing failed and the rank badge simply never appeared. Order is the
|
|
||||||
# behaviour here, so these tests assert it directly.
|
|
||||||
# ------------------------------------------------------------------
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def _requested_endpoints(mock_get):
|
|
||||||
"""The endpoint names probed, in the order they were requested."""
|
|
||||||
return [call.args[0].rsplit("/", 1)[-1] for call in mock_get.call_args_list]
|
|
||||||
|
|
||||||
def test_professional_league_asks_standings_first(self):
|
|
||||||
payload = {"standings": []}
|
|
||||||
with patch.object(self.source.session, "get",
|
|
||||||
return_value=_mock_response(payload)) as mock_get:
|
|
||||||
result = self.source.fetch_standings("football", "nfl")
|
|
||||||
assert result == payload
|
|
||||||
assert self._requested_endpoints(mock_get) == ["standings"]
|
|
||||||
|
|
||||||
def test_college_league_asks_rankings_first(self):
|
|
||||||
poll = {"rankings": [{"name": "AP Top 25"}]}
|
|
||||||
with patch.object(self.source.session, "get",
|
|
||||||
return_value=_mock_response(poll)) as mock_get:
|
|
||||||
result = self.source.fetch_standings("football", "college-football")
|
|
||||||
assert result == poll
|
|
||||||
assert self._requested_endpoints(mock_get) == ["rankings"]
|
|
||||||
|
|
||||||
def test_rankings_200_without_a_poll_falls_through_to_standings(self):
|
|
||||||
"""A 200 is not the same as an answer.
|
|
||||||
|
|
||||||
This is the case the old code could not see: the endpoint responded,
|
|
||||||
so nothing raised, but the body carried no poll.
|
|
||||||
"""
|
|
||||||
empty_poll = _mock_response({"rankings": []})
|
|
||||||
table = _mock_response({"standings": [{"entries": []}]})
|
|
||||||
with patch.object(self.source.session, "get",
|
|
||||||
side_effect=[empty_poll, table]) as mock_get:
|
|
||||||
result = self.source.fetch_standings(
|
|
||||||
"basketball", "mens-college-basketball")
|
|
||||||
assert result == {"standings": [{"entries": []}]}
|
|
||||||
assert self._requested_endpoints(mock_get) == ["rankings", "standings"]
|
|
||||||
|
|
||||||
def test_404_on_the_first_endpoint_falls_through_quietly(self):
|
|
||||||
missing = _mock_response({}, status_code=404)
|
|
||||||
table = _mock_response({"standings": []})
|
|
||||||
with patch.object(self.source.session, "get",
|
|
||||||
side_effect=[missing, table]) as mock_get:
|
|
||||||
result = self.source.fetch_standings("baseball", "college-baseball")
|
|
||||||
assert result == {"standings": []}
|
|
||||||
assert self._requested_endpoints(mock_get) == ["rankings", "standings"]
|
|
||||||
|
|
||||||
def test_recovers_from_a_non_404_failure_on_the_first_endpoint(self):
|
|
||||||
table = _mock_response({"standings": [{"entries": []}]})
|
|
||||||
with patch.object(self.source.session, "get",
|
|
||||||
side_effect=[requests.ConnectionError("reset"), table]) as mock_get:
|
|
||||||
result = self.source.fetch_standings("football", "college-football")
|
|
||||||
assert result == {"standings": [{"entries": []}]}
|
|
||||||
assert self._requested_endpoints(mock_get) == ["rankings", "standings"]
|
|
||||||
|
|
||||||
def test_both_endpoints_failing_returns_empty(self):
|
|
||||||
with patch.object(self.source.session, "get",
|
|
||||||
side_effect=requests.ConnectionError("down")) as mock_get:
|
|
||||||
result = self.source.fetch_standings("football", "nfl")
|
|
||||||
assert result == {}
|
|
||||||
assert self._requested_endpoints(mock_get) == ["standings", "rankings"]
|
|
||||||
|
|
||||||
def test_a_non_object_payload_is_treated_as_a_miss(self):
|
|
||||||
odd = _mock_response(["not", "an", "object"])
|
|
||||||
table = _mock_response({"standings": []})
|
|
||||||
with patch.object(self.source.session, "get",
|
|
||||||
side_effect=[odd, table]) as mock_get:
|
|
||||||
result = self.source.fetch_standings("football", "college-football")
|
|
||||||
assert result == {"standings": []}
|
|
||||||
assert self._requested_endpoints(mock_get) == ["rankings", "standings"]
|
|
||||||
|
|
||||||
def test_a_bug_in_this_method_is_not_swallowed_as_a_failed_endpoint(self):
|
|
||||||
"""The guard for the narrowed handler.
|
|
||||||
|
|
||||||
An error raised while reading the payload used to be caught by the
|
|
||||||
endpoint handler and reported as 'no poll here', which would silently
|
|
||||||
drop rankings for a league that has them. It must surface instead.
|
|
||||||
"""
|
|
||||||
boom = Mock(spec=requests.Response)
|
|
||||||
boom.status_code = 200
|
|
||||||
boom.raise_for_status = Mock()
|
|
||||||
boom.json.side_effect = TypeError("a bug, not a network failure")
|
|
||||||
with patch.object(self.source.session, "get", return_value=boom):
|
|
||||||
with pytest.raises(TypeError):
|
|
||||||
self.source.fetch_standings("football", "nfl")
|
|
||||||
|
|
||||||
def test_base_url_set_correctly(self):
|
def test_base_url_set_correctly(self):
|
||||||
assert "espn.com" in self.source.base_url
|
assert "espn.com" in self.source.base_url
|
||||||
|
|
||||||
|
|||||||
@@ -1,120 +0,0 @@
|
|||||||
"""The generated sudoers rules must parse before they reach /etc/sudoers.d.
|
|
||||||
|
|
||||||
A malformed drop-in there makes sudo refuse every command for every user. On a
|
|
||||||
headless Pi that is unrecoverable without pulling the SD card, so both
|
|
||||||
installers run `visudo -c` on the file they generated before installing it.
|
|
||||||
|
|
||||||
The render test also gives us the check neither installer had: that the rules
|
|
||||||
they actually emit are valid sudoers syntax on a real Linux box.
|
|
||||||
"""
|
|
||||||
|
|
||||||
import os
|
|
||||||
import shutil
|
|
||||||
import subprocess
|
|
||||||
import sys
|
|
||||||
import tempfile
|
|
||||||
|
|
||||||
import pytest
|
|
||||||
|
|
||||||
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
||||||
FIRST_TIME = os.path.join(REPO_ROOT, "first_time_install.sh")
|
|
||||||
CONFIGURE = os.path.join(REPO_ROOT, "scripts", "install", "configure_web_sudo.sh")
|
|
||||||
|
|
||||||
VISUDO = shutil.which("visudo") or (
|
|
||||||
"/usr/sbin/visudo" if os.path.exists("/usr/sbin/visudo") else None
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def _read(path):
|
|
||||||
with open(path, "r", encoding="utf-8") as handle:
|
|
||||||
return handle.read()
|
|
||||||
|
|
||||||
|
|
||||||
def test_first_time_install_validates_before_installing():
|
|
||||||
body = _read(FIRST_TIME)
|
|
||||||
assert 'visudo -c -f "$SUDOERS_TMP"' in body
|
|
||||||
install = body.index('cp "$SUDOERS_TMP" "$SUDOERS_FILE"')
|
|
||||||
validate = body.index('visudo -c -f "$SUDOERS_TMP"')
|
|
||||||
assert validate < install, "the rules must be checked before they are installed"
|
|
||||||
|
|
||||||
|
|
||||||
def test_the_install_is_gated_on_the_check():
|
|
||||||
"""Checking and then installing anyway would be worse than not checking."""
|
|
||||||
body = _read(FIRST_TIME)
|
|
||||||
assert "SUDOERS_VALID=0" in body
|
|
||||||
gate = body.index('if [ "$SUDOERS_VALID" = "0" ]')
|
|
||||||
install = body.index('cp "$SUDOERS_TMP" "$SUDOERS_FILE"')
|
|
||||||
assert gate < install
|
|
||||||
|
|
||||||
|
|
||||||
def test_first_time_install_does_not_use_a_predictable_temp_file():
|
|
||||||
body = _read(FIRST_TIME)
|
|
||||||
assert "mktemp" in body
|
|
||||||
assert "> /tmp/ledmatrix_web_sudoers" not in body
|
|
||||||
assert ">> /tmp/ledmatrix_web_sudoers" not in body
|
|
||||||
|
|
||||||
|
|
||||||
def test_configure_web_sudo_validates_before_installing():
|
|
||||||
body = _read(CONFIGURE)
|
|
||||||
assert 'visudo -c -f "$TEMP_SUDOERS"' in body
|
|
||||||
install = body.index('cp "$TEMP_SUDOERS" /etc/sudoers.d/ledmatrix_web')
|
|
||||||
validate = body.index('visudo -c -f "$TEMP_SUDOERS"')
|
|
||||||
assert validate < install, "the rules must be checked before they are installed"
|
|
||||||
|
|
||||||
|
|
||||||
def _render_first_time_sudoers(project_root, user):
|
|
||||||
"""Run the installer's own sudoers heredoc with realistic values."""
|
|
||||||
body = _read(FIRST_TIME)
|
|
||||||
start = body.index("# Create sudoers content")
|
|
||||||
end = body.index("# Never install rules we have not parsed.")
|
|
||||||
block = body[start:end]
|
|
||||||
out = os.path.join(project_root, "rendered")
|
|
||||||
script = "\n".join(
|
|
||||||
[
|
|
||||||
"set -euo pipefail",
|
|
||||||
f"ACTUAL_USER={user}",
|
|
||||||
f"PROJECT_ROOT_DIR={project_root}",
|
|
||||||
'SUDOERS_TMP="$(mktemp)"',
|
|
||||||
"PYTHON_PATH=$(which python3)",
|
|
||||||
"SYSTEMCTL_PATH=/usr/bin/systemctl",
|
|
||||||
"REBOOT_PATH=/usr/sbin/reboot",
|
|
||||||
"POWEROFF_PATH=/usr/sbin/poweroff",
|
|
||||||
"BASH_PATH=$(which bash)",
|
|
||||||
"JOURNALCTL_PATH=/usr/bin/journalctl",
|
|
||||||
block,
|
|
||||||
f'cp "$SUDOERS_TMP" {out}',
|
|
||||||
]
|
|
||||||
)
|
|
||||||
subprocess.run(["bash", "-c", script], check=True)
|
|
||||||
return out
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.skipif(sys.platform == "win32", reason="visudo is POSIX only")
|
|
||||||
@pytest.mark.skipif(VISUDO is None, reason="visudo not installed")
|
|
||||||
def test_the_rules_the_installer_emits_actually_parse():
|
|
||||||
with tempfile.TemporaryDirectory() as tmp:
|
|
||||||
rendered = _render_first_time_sudoers(tmp, "ledmatrix")
|
|
||||||
os.chmod(rendered, 0o440)
|
|
||||||
result = subprocess.run(
|
|
||||||
[VISUDO, "-c", "-f", rendered], capture_output=True, text=True
|
|
||||||
)
|
|
||||||
assert result.returncode == 0, result.stdout + result.stderr
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.skipif(sys.platform == "win32", reason="visudo is POSIX only")
|
|
||||||
@pytest.mark.skipif(VISUDO is None, reason="visudo not installed")
|
|
||||||
def test_a_broken_rule_is_caught_rather_than_installed():
|
|
||||||
"""The guard is only worth having if visudo rejects what it should."""
|
|
||||||
with tempfile.TemporaryDirectory() as tmp:
|
|
||||||
rendered = _render_first_time_sudoers(tmp, "ledmatrix")
|
|
||||||
with open(rendered, "r", encoding="utf-8") as handle:
|
|
||||||
good = handle.read()
|
|
||||||
broken = os.path.join(tmp, "broken")
|
|
||||||
with open(broken, "w", encoding="utf-8") as handle:
|
|
||||||
# An empty command path is what an unset $BASH_PATH would produce.
|
|
||||||
handle.write(good + "\nledmatrix ALL=(ALL) NOPASSWD:\n")
|
|
||||||
os.chmod(broken, 0o440)
|
|
||||||
result = subprocess.run(
|
|
||||||
[VISUDO, "-c", "-f", broken], capture_output=True, text=True
|
|
||||||
)
|
|
||||||
assert result.returncode != 0
|
|
||||||
Reference in New Issue
Block a user