ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add log collection, script output capture, and timestamp fixes

Browse Source 
Log Collector: a new collect_log action reads a plaintext log file from a
known path and returns its tail for viewing and download. Rides the existing
endpoints.updateapp grant rather than inventing one, since collecting a log is
the same trust level as pushing an application. Collected logs land in their
own table with per-asset retention of 20 instead of overloading the command
audit row, and the agent refuses UNC, traversal, wildcards, directories and
non-text content independently of the server check.

collect_log is POLICY_EXEMPT. Every existing Configuration Policy predates the
action and so cannot name it, and the intersection in ALLOWED_ACTIONS_FOR drops
anything unnamed - so every policied endpoint refused log collection outright,
including for admins, who saw a permission-shaped error for a grant they held.
Exemption from policy is not exemption from permission: the elevated grant
still applies, and a regression test asserts both halves so a future edit
cannot quietly loosen the grant while "fixing" the policy gate.

Script output capture: successful runs can return output, gated per script by
captureoutput. Failures always returned it; success never did, which left a
green badge as the only evidence a run produced.

Timestamps: Activity and Operator Reports rendered date-only because the shared
formatter's 'date' branch drops the clock time. Run Scripts additionally
labels its timestamp, since a bare time meant "completed" on a terminal row but
"queued" on an in-flight one - indistinguishable, and the wrong reading.

Agent 1.16.0. Minor: collect_log is a new action name, so an older agent asked
for it fails closed on the unknown-action default.
This commit is contained in:
benjamin.baker
2026-09-09 19:23:43 -04:00
parent 494b837bae
commit b49fb60c0c
19 changed files with 2207 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

129
tests/test_core/test_script_captureoutput.py Normal file
View File

@@ -0,0 +1,129 @@
"""The per-script "return output on success" flag.
A failed run has always sent its output back, because a failure has to be
diagnosable from the UI. A successful run has not, and that is the behaviour
being made opt-in here rather than simply reversed: script output contains
whatever the body printed, and persisting that server-side for every routine
green run is exposure nobody asked for.
The load-bearing property is the default. Every script authored before this
field existed ran under "success stays silent", so an absent field must never
be read as consent.
"""
import pytest
from werkzeug.security import generate_password_hash
from shopdb.core.models import Permission
SCRIPT_GRANT = ('endpoints.runscript', 'Run staged scripts', 'endpoints')
@pytest.fixture
def headers(client, db):
from shopdb.core.models import User
Permission.seed_entries([SCRIPT_GRANT])
db.session.commit()
user = User(
username='captureauthor',
email='captureauthor@test.local',
passwordhash=generate_password_hash('testpass'),
)
user.directpermissions.append(
Permission.query.filter_by(name='endpoints.runscript').one())
db.session.add(user)
db.session.commit()
login = client.post('/api/auth/login',
json={'username': 'captureauthor', 'password': 'testpass'})
assert login.status_code == 200, login.get_json()
return {'Authorization': f"Bearer {login.get_json()['data']['access_token']}"}
def _create(client, headers, name, **extra):
payload = {'scriptname': name}
payload.update(extra)
return client.post('/api/scripts', headers=headers, json=payload)
def test_a_script_without_the_flag_withholds_output(client, headers):
"""The load-bearing test. An omitted field is not consent to persist a
successful run's output server-side."""
created = _create(client, headers, 'no-capture')
assert created.status_code == 201, created.get_json()
assert created.get_json()['data']['captureoutput'] is False
def test_the_flag_can_be_turned_on(client, headers):
created = _create(client, headers, 'chatty-one', captureoutput=True)
assert created.status_code == 201, created.get_json()
assert created.get_json()['data']['captureoutput'] is True
@pytest.mark.parametrize('sent,expected', [
('true', True),
('false', False),
(True, True),
(False, False),
])
def test_form_style_values_are_accepted(client, headers, sent, expected):
"""The field arrives from a checkbox, which may send a string."""
created = _create(client, headers, f'form-{sent}-{expected}',
captureoutput=sent)
assert created.status_code == 201, created.get_json()
assert created.get_json()['data']['captureoutput'] is expected
def test_a_nonsense_value_is_rejected(client, headers):
"""Rejected rather than coerced: silently reading garbage as False would
hide a broken client, and as True would leak output nobody agreed to."""
created = _create(client, headers, 'bad-capture', captureoutput='maybe')
assert created.status_code == 400
assert 'captureoutput' in created.get_json()['data']['error']['message']
def test_the_flag_can_be_changed_on_an_existing_script(client, headers):
scriptid = _create(client, headers, 'toggle-me',
captureoutput=True).get_json()['data']['scriptid']
updated = client.put(f'/api/scripts/{scriptid}', headers=headers,
json={'captureoutput': False})
assert updated.status_code == 200, updated.get_json()
assert updated.get_json()['data']['captureoutput'] is False
def test_an_unrelated_update_leaves_the_flag_alone(client, headers):
"""Editing a description must not quietly revert an author's decision."""
scriptid = _create(client, headers, 'keep-my-flag',
captureoutput=True).get_json()['data']['scriptid']
updated = client.put(f'/api/scripts/{scriptid}', headers=headers,
json={'scriptdescription': 'now documented'})
assert updated.get_json()['data']['captureoutput'] is True
def test_the_delivery_descriptor_carries_the_flag(app, client, db, headers):
"""The agent cannot honour a decision it is never told about."""
from plugins.endpoints.api.routes import _resolve_run_script
with app.app_context():
from shopdb.core.models import Script
script = Script(scriptname='delivered', filename='script-1.ps1',
sha256='a' * 64, runcontext='system',
captureoutput=True)
db.session.add(script)
db.session.commit()
descriptor, err = _resolve_run_script(f'script:{script.scriptid}')
assert err is None
assert descriptor['captureoutput'] is True

20
tests/test_plugins/test_endpoints_plugin.py
View File

@@ -196,6 +196,26 @@ def test_collect_now_survives_a_policy_that_never_mentions_it():
assert elevated_permission_for('collect_now') is None
def test_collect_log_survives_a_policy_that_never_mentions_it():
"""Log collection must not be narrowed away by an unrelated policy.
Same intersection trap as collect_now, but with a sharper failure: every
existing policy predates this action and so cannot name it, meaning any
endpoint carrying a policy refused log collection outright - including for
an admin, who reads "requires permission" and goes looking at a grant they
already hold. Logs are also what an operator reaches for when an endpoint
is already misbehaving, so a policy gate here withholds the evidence
exactly when it is wanted.
"""
from plugins.endpoints.commands import POLICY_EXEMPT_ACTIONS
assert 'collect_log' in POLICY_EXEMPT_ACTIONS
# Exemption from POLICY is not exemption from PERMISSION. Unlike
# collect_now this one DOES carry an elevated grant, so the assertion is
# that the grant survived the exemption rather than being loosened by it.
assert elevated_permission_for('collect_log') == 'endpoints.updateapp'
def test_run_script_and_update_app_are_allowlisted_delivery_actions():
"""The two software-delivery actions are allowlisted with a targetkind so
the enqueue route accepts them and carries their subject on the wire:

200
tests/test_plugins/test_endpoints_script_runs.py Normal file
View File

@@ -0,0 +1,200 @@
"""The Run Scripts tab's history feed.
A successful script produces a green badge and a one-line message; nothing in
that tells an operator whether the script did any work. These tests pin the
contract the tab relies on: only script runs, newest first, capped, and
carrying the agent's structured detail so the exit code and run context are
visible without opening the database.
"""
import json
from werkzeug.security import generate_password_hash
def _endpoint(db, hostname='scriptbox'):
"""An Asset + Computer pair with a reported state, as the routes expect."""
from shopdb.api import Asset, AssetType
from plugins.computers.models import Computer
from plugins.endpoints.models import EndpointState
assettype = AssetType.query.filter_by(assettype='computer').first()
if assettype is None:
assettype = AssetType(assettype='computer', pluginname='computers',
tablename='computers')
db.session.add(assettype)
db.session.flush()
asset = Asset(assetnumber=hostname, name=hostname,
assettypeid=assettype.assettypeid, isactive=True)
db.session.add(asset)
db.session.flush()
db.session.add(Computer(assetid=asset.assetid, hostname=hostname))
db.session.add(EndpointState(assetid=asset.assetid, hostname=hostname))
db.session.commit()
return asset.assetid
def _viewer(db, client, username='scriptviewer'):
from shopdb.core.models import User, Role, Permission
Permission.seed()
db.session.flush()
role = Role.query.filter_by(rolename='scriptviewerrole').first()
if role is None:
role = Role(rolename='scriptviewerrole')
db.session.add(role)
db.session.flush()
role.permissions = Permission.query.filter(
Permission.name.in_(['endpoints.view'])).all()
user = User(username=username, email=username + '@test.local',
passwordhash=generate_password_hash('testpass'))
user.roles.append(role)
db.session.add(user)
db.session.commit()
response = client.post('/api/auth/login',
json={'username': username, 'password': 'testpass'})
token = response.get_json()['data']['access_token']
return {'Authorization': f'Bearer {token}'}
def _script(db, name='test-open-notepad'):
from shopdb.core.models import Script
script = Script(scriptname=name, runcontext='system')
db.session.add(script)
db.session.commit()
return script.scriptid
def _command(db, assetid, target, createdat, action='run_script',
status='succeeded', resultjson=None, resultmessage=None):
from plugins.endpoints.models import EndpointCommand
command = EndpointCommand(assetid=assetid, action=action, target=target,
status=status, resultmessage=resultmessage,
resultjson=resultjson, createdat=createdat)
db.session.add(command)
db.session.commit()
return command.endpointcommandid
def test_only_script_runs_are_returned(app, client, db):
"""A restart sitting in the same feed is not a script run.
The tab asks a narrower question than the activity panel, so filtering has
to happen server-side; a mixed feed sliced client-side would quietly show
fewer than five script runs whenever other commands interleave.
"""
from datetime import datetime
with app.app_context():
assetid = _endpoint(db)
scriptid = _script(db)
headers = _viewer(db, client)
_command(db, assetid, f'script:{scriptid}', datetime(2026, 1, 1, 10, 0))
_command(db, assetid, 'restart', datetime(2026, 1, 1, 11, 0),
action='restart')
response = client.get(f'/api/endpoints/{assetid}/scriptruns',
headers=headers)
rows = response.get_json()['data']
assert response.status_code == 200
assert [row['action'] for row in rows] == ['run_script']
def test_newest_first_and_capped_at_five(app, client, db):
"""Six runs, five rows, most recent at the top."""
from datetime import datetime
with app.app_context():
assetid = _endpoint(db)
scriptid = _script(db)
headers = _viewer(db, client)
for hour in range(6):
_command(db, assetid, f'script:{scriptid}',
datetime(2026, 1, 1, 10 + hour, 0))
response = client.get(f'/api/endpoints/{assetid}/scriptruns',
headers=headers)
rows = response.get_json()['data']
assert len(rows) == 5
stamps = [row['createdat'] for row in rows]
assert stamps == sorted(stamps, reverse=True)
def test_agent_detail_is_exposed(app, client, db):
"""The exit code and run context are the evidence behind a green badge."""
from datetime import datetime
with app.app_context():
assetid = _endpoint(db)
scriptid = _script(db)
headers = _viewer(db, client)
_command(db, assetid, f'script:{scriptid}', datetime(2026, 1, 1, 10, 0),
resultmessage='Script completed (system context)',
resultjson=json.dumps({'kind': 'script', 'exitcode': 0,
'runcontext': 'system'}))
response = client.get(f'/api/endpoints/{assetid}/scriptruns',
headers=headers)
row = response.get_json()['data'][0]
assert row['resultdetail']['exitcode'] == 0
assert row['resultdetail']['runcontext'] == 'system'
def test_the_script_name_is_resolved(app, client, db):
"""'script:5' is the wire target, not what the operator picked."""
from datetime import datetime
with app.app_context():
assetid = _endpoint(db)
scriptid = _script(db, name='rebuild-shopfloor-user')
headers = _viewer(db, client)
_command(db, assetid, f'script:{scriptid}', datetime(2026, 1, 1, 10, 0))
response = client.get(f'/api/endpoints/{assetid}/scriptruns',
headers=headers)
row = response.get_json()['data'][0]
assert row['scriptname'] == 'rebuild-shopfloor-user'
def test_malformed_detail_does_not_break_the_list(app, client, db):
"""resultjson is agent-supplied text and can arrive truncated.
A diagnostics panel failing to parse must degrade to "no detail", not take
down the history that surrounds it.
"""
from datetime import datetime
with app.app_context():
assetid = _endpoint(db)
scriptid = _script(db)
headers = _viewer(db, client)
_command(db, assetid, f'script:{scriptid}', datetime(2026, 1, 1, 10, 0),
resultjson='{"kind": "script", "exitcode":')
response = client.get(f'/api/endpoints/{assetid}/scriptruns',
headers=headers)
assert response.status_code == 200
assert response.get_json()['data'][0]['resultdetail'] is None
def test_view_permission_is_required(app, client, db):
"""History reveals what has been run on a machine, so it is gated."""
from datetime import datetime
with app.app_context():
assetid = _endpoint(db)
scriptid = _script(db)
_command(db, assetid, f'script:{scriptid}', datetime(2026, 1, 1, 10, 0))
response = client.get(f'/api/endpoints/{assetid}/scriptruns')
assert response.status_code in (401, 403)