sensitive-logging-audit

1from __future__ import annotations2 3import unittest4from pathlib import Path5from tempfile import TemporaryDirectory6 7from inventory_logging import collect_source_files, inventory_source, summarize8 9 10class InventoryTests(unittest.TestCase):11    def test_collects_direct_logging_calls_without_certifying_receivers(self) -> None:12        candidates = inventory_source(13            """14from logging import error15 16logger.debug("ready")17logger.error("failed: %s", secret)18error(secret)19task.exception()20"""21        )22 23        self.assertEqual(24            [(item.kind, item.method) for item in candidates],25            [26                ("logging-call-candidate", "debug"),27                ("logging-call-candidate", "error"),28                ("logging-call-candidate", "error"),29                ("logging-call-candidate", "exception"),30            ],31        )32 33    def test_collects_policy_helpers_without_claiming_their_callers_are_safe(self) -> None:34        candidates = inventory_source(35            """36from agents.logger import log_model_action_error37 38log_model_action_error(logger, "failed", error)39agents.logger.log_model_and_tool_action_warning(logger, "failed", error)40"""41        )42 43        self.assertEqual(44            [(item.kind, item.method) for item in candidates],45            [46                ("policy-helper-call", "log_model_action_error"),47                ("policy-helper-call", "log_model_and_tool_action_warning"),48            ],49        )50 51    def test_collects_raw_output_method_names(self) -> None:52        candidates = inventory_source(53            """54import os55import pprint56import sys57import traceback58import warnings59 60print(secret)61pprint.pp(secret)62warnings.warn(secret)63sys.stderr.buffer.write(secret_bytes)64sys.stdout.writelines([secret])65traceback.print_exception(error)66os.write(2, secret_bytes)67"""68        )69 70        self.assertEqual(71            [(item.kind, item.method) for item in candidates],72            [73                ("raw-output-call-candidate", "print"),74                ("raw-output-call-candidate", "pp"),75                ("raw-output-call-candidate", "warn"),76                ("raw-output-call-candidate", "write"),77                ("raw-output-call-candidate", "writelines"),78                ("raw-output-call-candidate", "print_exception"),79                ("raw-output-call-candidate", "write"),80            ],81        )82 83    def test_collects_constant_getattr_sink_selections(self) -> None:84        candidates = inventory_source(85            """86emit = getattr(logger, "error")87writer = builtins.getattr(stream, "write")88ignored = getattr(logger, method_name)89"""90        )91 92        self.assertEqual(93            [(item.kind, item.method) for item in candidates],94            [95                ("getattr-sink-candidate", "error"),96                ("getattr-sink-candidate", "write"),97            ],98        )99 100    def test_collects_obvious_logging_callbacks(self) -> None:101        candidates = inventory_source(102            """103register(log.warning)104register(on_error=service.error)105register(result=request.error)106"""107        )108 109        self.assertEqual(110            [(item.kind, item.method, item.call) for item in candidates],111            [112                ("logging-callback-candidate", "warning", "log.warning"),113                ("logging-callback-candidate", "error", "service.error"),114            ],115        )116 117    def test_keeps_the_output_schema_free_of_security_certification(self) -> None:118        candidate = inventory_source('logger.error("failed", secret)')[0].to_dict()119 120        self.assertEqual(121            set(candidate),122            {123                "fingerprint",124                "file",125                "line",126                "column",127                "kind",128                "method",129                "context",130                "call",131                "reason",132            },133        )134        self.assertNotIn("policy", candidate)135        self.assertNotIn("safe", candidate)136 137    def test_reports_enclosing_scope_as_review_context(self) -> None:138        candidate = inventory_source(139            """140class Worker:141    def report(self):142        logger.error(secret)143"""144        )[0]145 146        self.assertEqual(candidate.context, "class:Worker>function:report")147 148    def test_does_not_claim_to_follow_assignment_aliases(self) -> None:149        candidates = inventory_source(150            """151emit = logger.error152emit(secret)153"""154        )155 156        self.assertEqual(candidates, [])157 158    def test_summary_counts_only_syntactic_candidate_categories(self) -> None:159        candidates = inventory_source(160            """161logger.error(secret)162print(secret)163log_tool_action_error(logger, "failed", error)164register(on_error=service.error)165getattr(logger, "warning")166"""167        )168 169        self.assertEqual(170            summarize(candidates),171            {172                "totalCandidates": 5,173                "loggingCalls": 1,174                "rawOutputCalls": 1,175                "policyHelperCalls": 1,176                "getattrSelections": 1,177                "callbackReferences": 1,178            },179        )180 181    def test_collect_source_files_filters_hidden_children_relative_to_root(self) -> None:182        with TemporaryDirectory(prefix=".hidden-parent-") as directory:183            root = Path(directory) / "scan"184            root.mkdir()185            visible = root / "visible.py"186            visible.write_text("print('visible')\n")187            hidden = root / ".cache"188            hidden.mkdir()189            (hidden / "hidden.py").write_text("print('hidden')\n")190 191            self.assertEqual(collect_source_files([root]), [visible.resolve()])192 193 194if __name__ == "__main__":195    unittest.main()196