mirror of
https://github.com/Donkie/Spoolman.git
synced 2026-10-22 10:02:17 -05:00
Donkie
7d683eb246
Two lower-severity findings from the audit (todo.md tasks 6 and 7). A spreadsheet executes a cell that begins with =, +, -, @, tab or CR, so a vendor named `=cmd|' /C calc'!A0` runs when someone opens the CSV export (CWE-1236). Such cells are now prefixed with a single quote, which spreadsheets read as "this is literal text" and do not display. Only strings are escaped -- numbers reach the writer as numeric types, so a negative number cannot be mistaken for a formula and must not grow a stray quote. Header names need no escaping: they are fixed attribute names or extra-field keys, which are already constrained to ^[a-z0-9_]+$. Exports also gained Content-Disposition, so they download under a sensible name rather than rendering in the browser. Extra-field values had no size limit and the columns are Text(), so a single request could store megabytes: unbounded database growth, and the resulting websocket event overflowed the 1 MiB default frame limit, which killed live updates for every connected client. Values are now capped at the same 65535 characters as settings. The number of extra fields per entity is capped too -- the per-entity key count was already bounded, since unknown keys are rejected, but the registry itself could be grown without limit, and it is cached in memory and embedded in every entity response. Finally, a setting larger than the column limit returned 500; the ValueError is now caught and returned as a 400. Verified end-to-end: a 2 MB extra value is refused on both create and update, an oversized setting returns 400 and leaves the setting unset, normal values still work, and a vendor named `=cmd|' /C calc'!A0` exports as `'=cmd|' /C calc'!A0` while the JSON export keeps the raw name.
91 lines
3.6 KiB
Python
91 lines
3.6 KiB
Python
"""Functionality for exporting data in various format."""
|
|
|
|
import asyncio
|
|
import csv
|
|
import json
|
|
from collections.abc import Iterable
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from spoolman.database import models
|
|
|
|
if TYPE_CHECKING:
|
|
from _typeshed import SupportsWrite
|
|
|
|
banned_attrs = {"awaitable_attrs", "metadata", "registry", "spools", "filaments"}
|
|
|
|
# A spreadsheet treats a cell starting with any of these as a formula, so a vendor named
|
|
# `=cmd|' /C calc'!A0` executes when the export is opened (CWE-1236). Prefixing with a single
|
|
# quote makes the spreadsheet read it as literal text; the quote is not part of the value.
|
|
FORMULA_PREFIXES = ("=", "+", "-", "@", "\t", "\r")
|
|
|
|
|
|
def escape_csv_value(value: Any) -> Any: # noqa: ANN401
|
|
"""Neutralize a value that a spreadsheet would otherwise interpret as a formula.
|
|
|
|
Only strings are escaped. Numbers reach the writer as numeric types, so a negative number
|
|
cannot be mistaken for a formula and must not grow a stray quote.
|
|
|
|
Args:
|
|
value: The cell value.
|
|
|
|
Returns:
|
|
Any: The value, prefixed with a single quote if it would be read as a formula.
|
|
|
|
"""
|
|
if isinstance(value, str) and value.startswith(FORMULA_PREFIXES):
|
|
return "'" + value
|
|
return value
|
|
|
|
|
|
async def flatten_sqlalchemy_object(obj: models.Base, parent_key: str = "", sep: str = ".") -> dict[str, Any]:
|
|
"""Recursively flattens a SQLAlchemy object into a dictionary with dot-separated keys."""
|
|
fields = {}
|
|
for attr in dir(obj):
|
|
# Check if the attribute is a column or a relationship
|
|
if not attr.startswith("_") and attr not in banned_attrs:
|
|
value = await getattr(obj.awaitable_attrs, attr)
|
|
|
|
if attr == "extra":
|
|
# Handle extra fields
|
|
for v in value:
|
|
fields[f"{parent_key}extra.{v.key}"] = v.value
|
|
continue
|
|
|
|
# Handle nested SQLAlchemy objects
|
|
if isinstance(value, models.Base):
|
|
nested_fields = await flatten_sqlalchemy_object(value, f"{parent_key}{attr}{sep}", sep=sep)
|
|
fields.update(nested_fields)
|
|
else:
|
|
# Use only columns and simple data types
|
|
fields[f"{parent_key}{attr}"] = value
|
|
return fields
|
|
|
|
|
|
async def dump_as_csv(sqlalchemy_objects: Iterable[models.Base], writer: "SupportsWrite[str]") -> None:
|
|
"""Export a list of objects as CSV to a writer. Nested objects are flattened with dot-separated keys."""
|
|
# Flatten each object and get all column names
|
|
all_flattened = await asyncio.gather(*[flatten_sqlalchemy_object(obj) for obj in sqlalchemy_objects])
|
|
|
|
# Collect all unique headers across flattened objects
|
|
headers = set()
|
|
for flattened_obj in all_flattened:
|
|
headers.update(flattened_obj.keys())
|
|
|
|
headers = sorted(headers) # Sort headers for consistent column ordering
|
|
|
|
# Write to CSV. Header names need no escaping: they are either fixed attribute names or
|
|
# extra-field keys, which are constrained to ^[a-z0-9_]+$.
|
|
csv_writer = csv.DictWriter(writer, fieldnames=headers)
|
|
csv_writer.writeheader()
|
|
for flattened_obj in all_flattened:
|
|
csv_writer.writerow({key: escape_csv_value(value) for key, value in flattened_obj.items()})
|
|
|
|
|
|
async def dump_as_json(sqlalchemy_objects: Iterable[models.Base], writer: "SupportsWrite[str]") -> None:
|
|
"""Export a list of objects as JSON to a writer. Nested objects are flattened with dot-separated keys."""
|
|
# Flatten each object and get all column names
|
|
all_flattened = await asyncio.gather(*[flatten_sqlalchemy_object(obj) for obj in sqlalchemy_objects])
|
|
|
|
# Write to JSON
|
|
json.dump(all_flattened, writer, default=str)
|