fix: harden report publication and serialization

README.md

@@ -96,3 +96,6 @@ Dual licensed: [AGPL-3.0](LICENSE), including commercial use that meets its term
   <a href="https://complete.tech"><img src="assets/logo.png" alt="CompleteTech LLC" width="160"></a><br>
   <sub>Built by <a href="https://complete.tech">CompleteTech LLC</a> · Innovation at Every Integration</sub>
 </p>
+
+An interruption after a complete report is published preserves that report and exits 130.
+CSV rejects NUL characters before writing a report.

node2shell_check.py

@@ -8,6 +8,7 @@
 import argparse
 import base64
 import csv
+import io
 import html
 import json
 import math
@@ -678,6 +679,8 @@ def positive_float(value):
 def csv_safe(value):
     if isinstance(value, list):
         value = " | ".join(value)
+    if isinstance(value, str) and "\x00" in value:
+        raise ValueError("CSV values cannot contain NUL characters")
     # Prevent spreadsheet formulas from untrusted page titles / response fields.
     if isinstance(value, str) and value.lstrip().startswith(("=", "+", "-", "@")):
         return "'" + value
@@ -820,7 +823,9 @@ def write_report(output, rows, *, format="csv", logo=DEFAULT_LOGO, brand="Comple
 def save_report(path, rows, *, force=False, **options):
     """Publish a complete file atomically, refusing an existing path by default."""
     if path is None:
-        write_report(sys.stdout, rows, **options)
+        rendered = io.StringIO(newline="")
+        write_report(rendered, rows, **options)
+        sys.stdout.write(rendered.getvalue())
         return
     path = Path(path)
     temporary = None
@@ -910,7 +915,10 @@ def main(argv=None) -> int:
         print("Interrupted; preserving completed results.", file=sys.stderr)
     try:
         save_report(args.output, rows, force=args.force, format=args.format, logo=logo, brand=args.brand, complete=complete)
-    except (OSError, ValueError) as exc:
+    except KeyboardInterrupt:
+        print("Interrupted while writing the report; output may be incomplete.", file=sys.stderr)
+        return 130
+    except (OSError, ValueError, csv.Error) as exc:
         print(f"Cannot write report: {exc}", file=sys.stderr)
         return 2
     if not complete:

test_publication_boundaries.py

@@ -0,0 +1,50 @@
+"""Portable CSV and single-output publication regressions; offline fixtures only."""
+import contextlib
+import csv
+import io
+import json
+from pathlib import Path
+import tempfile
+import unittest
+from unittest import mock
+import node2shell_check as c
+class CsvPublicationTests(unittest.TestCase):
+    def run_cli(self,*args):
+        with contextlib.redirect_stdout(io.StringIO()) as out,contextlib.redirect_stderr(io.StringIO()) as err:
+            code=c.main(list(args))
+        return code,out.getvalue(),err.getvalue()
+    def test_nul_inventory_csv_rejects_before_publication(self):
+        with tempfile.TemporaryDirectory() as directory:
+            inventory=Path(directory)/'inventory.json'
+            inventory.write_text(json.dumps({'schema_version':1,'runtime':'node','node_version':'24.22.1','label':'example\x00asset'}))
+            for mode in ['stdout','new','force']:
+                target=Path(directory)/(mode+'.csv');extra=[]
+                if mode!='stdout':extra=['--output',str(target)]
+                if mode=='force':target.write_text('KEEP');extra+=['--force']
+                code,out,err=self.run_cli('--inventory',str(inventory),'--format','csv',*extra)
+                self.assertEqual((2,''),(code,out));self.assertNotIn('Traceback',err)
+                if mode=='force':self.assertEqual('KEEP',target.read_text())
+                else:self.assertFalse(target.exists())
+            code,out,_=self.run_cli('--inventory',str(inventory),'--format','json')
+            self.assertEqual(0,code);self.assertEqual(1,len(json.loads(out)['results']))
+    def test_bare_carriage_return_in_inventory_keeps_one_csv_record(self):
+        with tempfile.TemporaryDirectory() as directory:
+            inventory=Path(directory)/'inventory.json'
+            inventory.write_text(json.dumps({'schema_version':1,'runtime':'node','node_version':'24.22.1','label':'example\r=1+1'}))
+            code,out,_=self.run_cli('--inventory',str(inventory),'--format','csv')
+            self.assertEqual(0,code)
+            rows=list(csv.DictReader(io.StringIO(out,newline='')))
+            self.assertEqual(1,len(rows));self.assertNotIn(None,rows[0])
+            self.assertTrue(any('\r=1+1' in value for value in rows[0].values()))
+    def test_partial_stdout_interrupt_has_one_write_and_exit130(self):
+        class Sink:
+            calls=0
+            def write(self,data):self.calls+=1;raise KeyboardInterrupt
+        sink=Sink()
+        with contextlib.redirect_stdout(sink),contextlib.redirect_stderr(io.StringIO()):
+            escaped=False
+            try:code=c.main(['--node-version','24.22.1','--format','json'])
+            except KeyboardInterrupt:escaped=True;code=130
+        self.assertFalse(escaped,'main must handle its publication interruption')
+        self.assertEqual(130,code);self.assertEqual(1,sink.calls)
+if __name__=='__main__':unittest.main()