Compare commits

..
Author SHA1 Message Date
SAPHID 8a904c95e4 Escape power report CSV metadata for spreadsheets 2026-10-24 07:19:40 +00:00
4 changed files with 30 additions and 7 deletions

No files matched your search

+3
View File
@@ -2,6 +2,9 @@
## Unreleased
- Treat headset aliases and other CSV metadata as plain text in power and heat
exports, so spreadsheet formula prefixes cannot run when a report is opened.
- Add desktop Tools → Power & heat report: bounded read-only one/five-minute
observations with progress, stop and HTML/JSON/CSV downloads. Received samples
survive stop, disconnect and headset switch, labeled incomplete and tied to the
+3 -1
View File
@@ -94,7 +94,9 @@ in your browser to view the offline table and its embedded JSON/CSV downloads.
Snapshots taken before completion explicitly say incomplete. Desktop CSV includes
`collection.state`, `collection.requestedSamples` and `collection.receivedSamples`
metadata rows with the original headset alias in the source column, even when no
sensor sample arrived. JSON records the alias and device ID; HTML names the alias.
sensor sample arrived. CSV text fields, including the alias, are escaped if they
begin with a spreadsheet formula prefix. JSON records the alias and device ID;
HTML names the alias.
Only the latest observation is retained in server memory, including partial
samples after stop or disconnect. **Download before starting another observation,
+10 -6
View File
@@ -141,6 +141,13 @@ def collect(status, count, interval, emit=None):
def csv_samples(report):
# CSV quoting alone does not prevent spreadsheet formulas in text fields.
def cell(value):
if isinstance(value, str) and (value.startswith(('\t', '\r', '\n')) or
value.lstrip(' \t\r\n').startswith(('=', '+', '-', '@'))):
return "'" + value
return value
output = io.StringIO(newline='')
writer = csv.writer(output)
writer.writerow(('time', 'elapsedSeconds', 'metric', 'value', 'unit', 'source', 'state'))
@@ -148,17 +155,14 @@ def csv_samples(report):
for key, value in (('collection.state', report['stopped'] or 'Completed'),
('collection.requestedSamples', report['requestedSamples']),
('collection.receivedSamples', len(report['samples']))):
writer.writerow(('', '', key, value, '', report['headset']['alias'],
'incomplete' if report['stopped'] else 'completed'))
writer.writerow([cell(v) for v in ('', '', key, value, '', report['headset']['alias'],
'incomplete' if report['stopped'] else 'completed')])
for sample in report['samples']:
for key, row in sample['readings'].items():
# csv quoting does not prevent spreadsheet formulas in labels.
def cell(value):
return "'" + value if isinstance(value, str) and value.startswith(('=', '+', '-', '@')) else value
writer.writerow([cell(v) for v in (sample['time'], sample['elapsedSeconds'], key,
row['value'], row['unit'], row['source'], row['state'])])
for group in sample['unavailableGroups']:
writer.writerow((sample['time'], sample['elapsedSeconds'], group, '', '', '', UNAVAILABLE))
writer.writerow([cell(v) for v in (sample['time'], sample['elapsedSeconds'], group, '', '', '', UNAVAILABLE)])
return output.getvalue()
+14
View File
@@ -2,6 +2,7 @@
import sandbox # noqa: F401
import base64
import contextlib
import csv
from html.parser import HTMLParser
import io
import json
@@ -43,6 +44,19 @@ class Report(unittest.TestCase):
with contextlib.redirect_stderr(io.StringIO()):
return power.stream_report([sys.executable, '-c', program], '', count, 1)
def test_csv_metadata_headset_alias_is_plain_text_in_spreadsheets(self):
for prefix in ('=', '+', '-', '@', ' =', '\t='):
with self.subTest(prefix=prefix):
alias = prefix + 'HYPERLINK("https://example.invalid")'
report = {'headset': {'alias': alias}, 'requestedSamples': 12,
'stopped': 'Stopped', 'samples': []}
rows = list(csv.reader(io.StringIO(power.csv_samples(report))))
self.assertEqual(len(rows), 4)
self.assertEqual([row[5] for row in rows[1:]], ["'" + alias] * 3)
report['headset']['alias'] = 'living-room-frame'
rows = list(csv.reader(io.StringIO(power.csv_samples(report))))
self.assertEqual([row[5] for row in rows[1:]], ['living-room-frame'] * 3)
def test_completed_stream_preserves_zero_unknown_units_and_downloads(self):
event = json.dumps({'sample': SAMPLE})
program = f'import sys; sys.stdin.read(); print({event!r}); print(\'{{"finished": {{"stopped": null}}}}\')'