Preserve reports when interrupted publication is indeterminate

README.md

@@ -57,3 +57,5 @@ Exit 0: completed run; 1: potentially affected with `--fail-on-finding`; 2: argu
 CSV `record_type=result` rows are asset observations. A zero-result report has a `record_type=summary` row carrying completeness, schema, rule date, CVE and `not_tested`; it is not an asset finding.
 
 CSV rejects NUL characters before writing a report. An interruption after a complete report is published preserves that report and exits 130.
+
+If an interrupt leaves file publication indeterminate, the checker exits 130 without retrying. Inspect the output before retrying; the diagnostic does not establish whether publication succeeded.

SHA256SUMS

@@ -1 +1 @@
-4dbc540413a840257a369d3e3fe5dbe02db07464ad72a8f3458f23c32c693c69  tinypool_options_rce_check.py
+52988c0a8ac271882ce5d4b39bd4ffa666f803022c76e9f9bf5f0cc41bf54580  tinypool_options_rce_check.py

test_publication_boundaries.py

@@ -12,6 +12,8 @@
 VERSION_FIELD = 'package_version'
 FIXED = '2.1.2'
 
+UNCERTAIN_ARGS = ['--version', FIXED]
+
 class PublicationBoundaryTests(unittest.TestCase):
     def run_cli(self,*args):
         with contextlib.redirect_stdout(io.StringIO()) as out,contextlib.redirect_stderr(io.StringIO()) as err:
@@ -76,4 +78,56 @@ def test_nul_csv_fails_before_stdout_new_or_forced_publication(self):
                     else:self.assertFalse(target.exists())
             code,out,_=self.run_cli('--inventory',str(inventory),'--format','json')
             self.assertEqual(0,code);self.assertEqual('example\x00asset',json.loads(out)['results'][0]['input_label'])
+    def test_indeterminate_publication_never_retries(self):
+        for fmt in ['json', 'csv', 'html']:
+            for force, attribute in [(False, 'link'), (True, 'replace')]:
+                for failure in ['metadata', 'second_interrupt']:
+                    with self.subTest(fmt=fmt, force=force, failure=failure), tempfile.TemporaryDirectory() as directory:
+                        target = Path(directory) / ('report.' + fmt)
+                        if force:
+                            target.write_text('old')
+                        original = getattr(c.os, attribute)
+                        original_stat = c.os.stat
+                        calls = []
+                        committed = []
+                        def publish(src, dst):
+                            calls.append(True)
+                            committed.append(Path(src).read_bytes())
+                            original(src, dst)
+                            raise KeyboardInterrupt
+                        def stat(path, *args, **kwargs):
+                            if kwargs.get('follow_symlinks') is False and Path(path) == target:
+                                if failure == 'second_interrupt':
+                                    raise KeyboardInterrupt
+                                raise OSError(5, 'synthetic metadata failure')
+                            return original_stat(path, *args, **kwargs)
+                        with mock.patch.object(c.os, attribute, publish), mock.patch.object(c.os, 'stat', stat):
+                            code, out, err = self.run_cli(*UNCERTAIN_ARGS, '--format', fmt, '--output', str(target), *(['--force'] if force else []))
+                        self.assertEqual(130, code)
+                        self.assertEqual('', out)
+                        self.assertEqual(1, len(calls))
+                        self.assertEqual(committed[0], target.read_bytes())
+                        self.assertNotIn('Traceback', err)
+                        self.assertNotIn('complete report was written', err)
+                        if failure == 'metadata':
+                            self.assertIn('indeterminate', err)
+                        self.assertEqual([target.name], sorted(x.name for x in Path(directory).iterdir()))
+
+    def test_definite_absence_allows_incomplete_recovery(self):
+        for force, attribute in [(False, 'link'), (True, 'replace')]:
+            with self.subTest(force=force), tempfile.TemporaryDirectory() as directory:
+                target = Path(directory) / 'report.json'
+                original = getattr(c.os, attribute)
+                calls = []
+                def publish(src, dst):
+                    calls.append(True)
+                    if len(calls) == 1:
+                        raise KeyboardInterrupt
+                    return original(src, dst)
+                with mock.patch.object(c.os, attribute, publish):
+                    code, _, _ = self.run_cli(*UNCERTAIN_ARGS, '--format', 'json', '--output', str(target), *(['--force'] if force else []))
+                self.assertEqual(130, code)
+                self.assertEqual(2, len(calls))
+                self.assertFalse(json.loads(target.read_text())['complete'])
+
 if __name__=='__main__':unittest.main()

tinypool_options_rce_check.py

@@ -469,6 +469,8 @@ def save_report(path, rows, *, force=False, state=None, **options):
             os.fsync(stream.fileno())
             staged_identity = os.fstat(stream.fileno())
         try:
+            if state is not None:
+                state["publication_started"] = True
             if force:
                 os.replace(temporary, path)
             else:
@@ -481,13 +483,19 @@ def save_report(path, rows, *, force=False, state=None, **options):
                                      f"({exc.strerror or type(exc).__name__}); choose another filesystem or explicitly use --force") from None
             state["published"] = True
         except KeyboardInterrupt:
-            # The OS can finish publication before Python delivers SIGINT.
-            # An existing path or equal bytes alone do not establish ownership.
+            # A second interrupt or metadata failure must not permit a retry.
             try:
-                if os.path.samestat(staged_identity, os.stat(path, follow_symlinks=False)):
-                    state["published"] = True
+                owned = os.path.samestat(staged_identity, os.stat(path, follow_symlinks=False))
+                if state is not None:
+                    if owned:
+                        state["published"] = True
+                    else:
+                        state["publication_started"] = False
+            except FileNotFoundError:
+                if state is not None:
+                    state["publication_started"] = False
             except OSError:
-                pass
+                pass  # Publication outcome remains indeterminate.
             raise
     finally:
         if temporary is not None:
@@ -587,6 +595,9 @@ def main(argv=None):
             if state["published"]:
                 print("Interrupted after the complete report was written; not retrying.", file=sys.stderr)
                 return 130
+            if state.get("publication_started"):
+                print("Interrupted; publication outcome indeterminate; inspect the output before retrying. No retry was attempted.", file=sys.stderr)
+                return 130
             if state["stdout_started"]:
                 print("Interrupted while writing to stdout; output may be truncated and no second report is written.", file=sys.stderr)
                 return 130