diff --git a/src/github_repo_auditor/portfolio_generation.py b/src/github_repo_auditor/portfolio_generation.py index 4abd183..f6f2ceb 100644 --- a/src/github_repo_auditor/portfolio_generation.py +++ b/src/github_repo_auditor/portfolio_generation.py @@ -106,6 +106,17 @@ def _parse_instant(value: object, *, field: str) -> datetime: return parsed.astimezone(UTC) +def _terminal_observed_in_order( + terminal_observed: datetime, produced_at: datetime, now: datetime +) -> bool: + """A terminal receipt may not be observed before the receipt it attests, nor + in the future. Terminal receipts carry a whole-second observed_at while + security receipts carry microseconds, so a terminal written in the same + second as the receipt (the normal case for a fast collector) compares at + second granularity; anything a full second earlier is still refused.""" + return produced_at.replace(microsecond=0) <= terminal_observed <= now.astimezone(UTC) + + def _instant_text(value: datetime) -> str: return value.astimezone(UTC).isoformat().replace("+00:00", "Z") @@ -322,7 +333,7 @@ def load_security_binding( terminal_observed = _parse_instant( terminal.get("observed_at"), field="security terminal observed_at" ) - if terminal_observed < produced_at or terminal_observed > now.astimezone(UTC): + if not _terminal_observed_in_order(terminal_observed, produced_at, now): raise PortfolioGenerationError("security terminal observation time is invalid") evidence = destination.get("evidence") if not isinstance(evidence, dict): diff --git a/tests/test_portfolio_generation.py b/tests/test_portfolio_generation.py index ab2ffe1..22b12b9 100644 --- a/tests/test_portfolio_generation.py +++ b/tests/test_portfolio_generation.py @@ -8,6 +8,7 @@ from pathlib import Path from github_repo_auditor.portfolio_generation import ( + _terminal_observed_in_order, ArtifactInput, PortfolioGenerationError, ProducerBinding, @@ -275,3 +276,19 @@ def test_rollback_and_roll_forward_swap_the_exact_pair(self): if __name__ == "__main__": unittest.main() + + +class TerminalObservationOrderTests(unittest.TestCase): + def test_same_second_terminal_is_accepted_and_earlier_second_is_refused(self) -> None: + from datetime import UTC, datetime + + produced = datetime(2026, 9, 2, 12, 8, 8, 576236, tzinfo=UTC) + now = datetime(2026, 9, 2, 12, 10, 0, tzinfo=UTC) + same_second = datetime(2026, 9, 2, 12, 8, 8, tzinfo=UTC) + earlier = datetime(2026, 9, 2, 12, 8, 7, tzinfo=UTC) + later = datetime(2026, 9, 2, 12, 8, 9, tzinfo=UTC) + future = datetime(2026, 9, 2, 12, 11, 0, tzinfo=UTC) + self.assertTrue(_terminal_observed_in_order(same_second, produced, now)) + self.assertTrue(_terminal_observed_in_order(later, produced, now)) + self.assertFalse(_terminal_observed_in_order(earlier, produced, now)) + self.assertFalse(_terminal_observed_in_order(future, produced, now))