Keep a share path inside the share, a UTC timestamp in UTC, and a public map public
Three confirmed findings from the 2026-09-03 run, each independent.
GE-Enforce scope names are directory names on the imaging share, and create_scope
checked only that the name was non-empty. "../escape" wrote a manifest one level
above the share root, "../../escape2" two, and "/tmp/anywhere" wherever it liked
- proven by writing the files and then deleting them. The exposure is the
permission split rather than the traversal itself: geenforce.manage creates the
scope and geenforce.publish writes the file, so a manage-only author could aim a
publisher's write at any path that process can reach. Names are now a single
safe path segment, checked at creation AND again at the write (rows predating the
check exist, and the write is the line that acts), with the resolved target
confirmed to sit under the share root.
Agent timestamps were converted with astimezone(tz=None), which means the
SERVER's local zone, not UTC. An agent reporting 17:26:16Z was stored as
13:26:16 on this EDT box and re-served labelled 'Z', while the same row's
server-written lastreportedat stayed true UTC - one row carrying two time bases
four hours apart. The dashboard's 48-hour stability window compares a real UTC
cutoff against these values, so events left it four hours early. Invisible on a
UTC server, which is how it survived, and nothing in the Dockerfile, the compose
files or the Windows deploy pins TZ. AGENT-CONTRACT.md mandates UTC on the wire,
so a naive value is now taken as UTC rather than guessed at.
The three /api/maplevels reads went back to optional-auth. ADR-017 rule 6, the
API inventory (auth "none"), the module's own docstring and the comment in
PrinterInstallerMap.vue all say they are public, because the printer installer
map runs before anyone logs in and needs a blueprint to draw on. afce2437 swept
165 genuinely-open GET routes behind require_permission and caught these in the
net, so the anonymous installer map has drawn no floor plan since. Restored
rather than amending the ADR: blueprint paths and level names are not secrets,
and the asset positions that go on top already render on public kiosk pages.
The writes stay gated, and a test proves it.
Registered in PUBLIC_ENDPOINTS with the reason. That gate checks a public route
is DECLARED public; the new test checks it actually answers, which is the half
that was broken.
Ten tests. All three verified by planting: restore the non-empty scope check and
eight fail, restore the local-zone conversion and two fail, put the permission
back on the level list and one fails.
Scope: product
This commit is contained in:
56
tests/test_core/test_maplevels_public.py
Normal file
56
tests/test_core/test_maplevels_public.py
Normal file
@@ -0,0 +1,56 @@
|
||||
"""The floor-map level reads answer a caller with no account.
|
||||
|
||||
ADR-017 rule 6 says so, docs/api-inventory.json declares auth "none" for all
|
||||
three, PrinterInstallerMap.vue carries a comment explaining why, and the module
|
||||
docstring in shopdb/core/api/maplevels.py says the reads are public. The code
|
||||
disagreed with all four: afce2437 swept 165 genuinely-open GET routes behind
|
||||
require_permission and caught these in the net, so the anonymous installer map
|
||||
drew no floor plan.
|
||||
|
||||
Nothing leaked - it failed closed - but a feature was off for a fortnight while
|
||||
every document said it was on. test_public_read_surface.py checks that a public
|
||||
route is REGISTERED as public; this checks that it actually answers, which is
|
||||
the part that was broken.
|
||||
"""
|
||||
|
||||
|
||||
def _level(db):
|
||||
from shopdb.core.models.maplevel import Building, MapLevel
|
||||
|
||||
building = Building(buildingname='Public Works', sortorder=1)
|
||||
db.session.add(building)
|
||||
db.session.flush()
|
||||
level = MapLevel(buildingid=building.buildingid, levelname='Ground',
|
||||
sortorder=1, blueprintlight='floorplan.svg')
|
||||
db.session.add(level)
|
||||
db.session.commit()
|
||||
return level.levelid
|
||||
|
||||
|
||||
def test_the_level_list_answers_an_anonymous_caller(client, db):
|
||||
levelid = _level(db)
|
||||
response = client.get('/api/maplevels')
|
||||
assert response.status_code == 200, response.get_json()
|
||||
payload = response.get_json()['data']
|
||||
names = [level['levelname']
|
||||
for building in payload['buildings']
|
||||
for level in building['levels']]
|
||||
assert 'Ground' in names
|
||||
assert levelid
|
||||
|
||||
|
||||
def test_one_level_answers_an_anonymous_caller(client, db):
|
||||
levelid = _level(db)
|
||||
response = client.get('/api/maplevels/%d' % levelid)
|
||||
assert response.status_code == 200, response.get_json()
|
||||
assert response.get_json()['data']['levelname'] == 'Ground'
|
||||
|
||||
|
||||
def test_writes_are_still_gated(client, db):
|
||||
"""The reads going back to public must not take the writes with them."""
|
||||
levelid = _level(db)
|
||||
assert client.post('/api/maplevels/buildings',
|
||||
json={'buildingname': 'Sneaky'}).status_code in (401, 403)
|
||||
assert client.put('/api/maplevels/%d' % levelid,
|
||||
json={'levelname': 'Renamed'}).status_code in (401, 403)
|
||||
assert client.delete('/api/maplevels/%d' % levelid).status_code in (401, 403)
|
||||
@@ -59,6 +59,13 @@ PUBLIC_ENDPOINTS = {
|
||||
'notifications.get_employee_recognitions',
|
||||
'slides.feed', 'slides.serve_image',
|
||||
|
||||
# The floor-map levels and their blueprints. ADR-017 rule 6: the printer
|
||||
# installer map runs optional-auth before anyone logs in and needs a
|
||||
# blueprint to draw on. Names and image paths, not asset data - and the
|
||||
# asset positions that go on top already render on public kiosk pages.
|
||||
'maplevels.list_levels', 'maplevels.get_level',
|
||||
'maplevels.serve_blueprint',
|
||||
|
||||
# Parts kiosk: one item by its code, and that item's picture.
|
||||
# kiosk_item takes no JWT but is NOT open: same kiosk flag + device token
|
||||
# as the write. Gating only the write would have left the enumeration
|
||||
|
||||
@@ -272,3 +272,66 @@ def test_fleet_list_route_resolves_without_trailing_slash():
|
||||
endpoint_slash, _ = adapter.match('/api/endpoints/', method='GET')
|
||||
assert endpoint_slash.endswith('list_endpoints')
|
||||
|
||||
|
||||
|
||||
# --- agent timestamps are UTC on the wire and UTC in the column --------------
|
||||
#
|
||||
# _parse_dt used astimezone(tz=None), which converts to the SERVER's local
|
||||
# zone. An agent reporting 17:26:16Z was stored as 13:26:16 on an EDT server
|
||||
# and re-served labelled 'Z', while the same row's server-written
|
||||
# lastreportedat stayed true UTC - one row, two time bases, four hours apart.
|
||||
# The dashboard's 48-hour stability window compares a real UTC cutoff against
|
||||
# these values, so events left it four hours early.
|
||||
#
|
||||
# It is invisible on a UTC server, which is how it survived, and nothing in the
|
||||
# Dockerfile, the compose files or the Windows deploy pins TZ.
|
||||
|
||||
def _in_zone(zonename, call):
|
||||
"""Run call() with the process in a named zone, then put it back."""
|
||||
import os
|
||||
import time
|
||||
|
||||
before = os.environ.get('TZ')
|
||||
os.environ['TZ'] = zonename
|
||||
time.tzset()
|
||||
try:
|
||||
return call()
|
||||
finally:
|
||||
if before is None:
|
||||
os.environ.pop('TZ', None)
|
||||
else:
|
||||
os.environ['TZ'] = before
|
||||
time.tzset()
|
||||
|
||||
|
||||
def test_a_utc_timestamp_survives_a_non_utc_server():
|
||||
from datetime import datetime
|
||||
|
||||
from plugins.endpoints.plugin import _parse_dt
|
||||
|
||||
parsed = _in_zone('America/New_York',
|
||||
lambda: _parse_dt('2026-09-03T17:26:16Z'))
|
||||
assert parsed == datetime(2026, 9, 3, 17, 26, 16), \
|
||||
'the wire value is UTC and the column is UTC; no local conversion'
|
||||
|
||||
|
||||
def test_an_offset_timestamp_is_converted_to_utc_not_to_server_local():
|
||||
from datetime import datetime
|
||||
|
||||
from plugins.endpoints.plugin import _parse_dt
|
||||
|
||||
parsed = _in_zone('America/New_York',
|
||||
lambda: _parse_dt('2026-09-03T17:26:16+05:00'))
|
||||
assert parsed == datetime(2026, 9, 3, 12, 26, 16)
|
||||
|
||||
|
||||
def test_a_naive_timestamp_is_taken_as_utc_already():
|
||||
"""AGENT-CONTRACT.md mandates UTC on the wire, so a value with no offset is
|
||||
UTC rather than a guess about the agent's zone."""
|
||||
from datetime import datetime
|
||||
|
||||
from plugins.endpoints.plugin import _parse_dt
|
||||
|
||||
parsed = _in_zone('Asia/Tokyo',
|
||||
lambda: _parse_dt('2026-09-03T17:26:16'))
|
||||
assert parsed == datetime(2026, 9, 3, 17, 26, 16)
|
||||
|
||||
@@ -63,3 +63,61 @@ def test_export_backs_up_existing(client, db, app, auth_headers, tmp_path):
|
||||
historydir = tmp_path / '_meta' / 'history'
|
||||
assert historydir.is_dir()
|
||||
assert any(historydir.iterdir())
|
||||
|
||||
|
||||
# --- a scope name is a directory on the share --------------------------------
|
||||
#
|
||||
# create_scope checked only that the name was non-empty, and the export did
|
||||
# os.path.join(shareroot, scopename, 'manifest.json'). So "../escape" wrote one
|
||||
# level above the share root and "/tmp/anywhere" wrote wherever it liked. The
|
||||
# exposure is the permission split: geenforce.manage creates the scope,
|
||||
# geenforce.publish writes the file, so a manage-only author could aim a
|
||||
# publisher's write at any path that process can reach.
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.mark.parametrize('name', [
|
||||
'../escape',
|
||||
'../../escape',
|
||||
'/tmp/absolute',
|
||||
'sub/dir',
|
||||
'back\\slash',
|
||||
'..',
|
||||
'.',
|
||||
])
|
||||
def test_a_traversing_scope_name_is_refused_at_creation(client, db,
|
||||
auth_headers, name):
|
||||
created = client.post('/api/geenforce/scopes',
|
||||
json={'scopename': name, 'phase': 'runtime'},
|
||||
headers=auth_headers)
|
||||
assert created.status_code == 400, (name, created.get_json())
|
||||
assert 'scopename' in created.get_json()['data']['error']['message']
|
||||
|
||||
|
||||
def test_a_normal_scope_name_is_still_accepted(client, db, auth_headers):
|
||||
"""The names sites actually use must keep working, or the guard is worse
|
||||
than the hole."""
|
||||
for name in ('common', 'display', 'gea-shopfloor-cmm', 'bay.3105_a'):
|
||||
created = client.post('/api/geenforce/scopes',
|
||||
json={'scopename': name, 'phase': 'runtime'},
|
||||
headers=auth_headers)
|
||||
assert created.status_code in (200, 201), (name, created.get_json())
|
||||
|
||||
|
||||
def test_the_export_refuses_a_traversing_name_even_from_the_database(
|
||||
app, db, tmp_path):
|
||||
"""Rows created before the validation existed are still out there, and the
|
||||
export is the line that actually writes."""
|
||||
with app.app_context():
|
||||
service.replace_scope_draft('../legacy-escape', 'runtime',
|
||||
{'Version': '2.6', 'Applications': []})
|
||||
service.publish_scope('../legacy-escape', 'runtime', notes='v1')
|
||||
service.db.session.commit()
|
||||
|
||||
with pytest.raises(ValueError) as refused:
|
||||
service.export_scope_to_share('../legacy-escape', 'runtime',
|
||||
str(tmp_path))
|
||||
assert 'safe directory name' in str(refused.value)
|
||||
|
||||
assert not (tmp_path.parent / 'legacy-escape').exists()
|
||||
|
||||
Reference in New Issue
Block a user