Harden scraper inputs and exports

tests/test_integration.py

@@ -284,6 +284,34 @@ class TestAccountLifecycle:
         assert loaded["continuous_scraping"]["run_all_tracked"] is True
 
 
+class TestInputSafety:
+    """Small safety checks for values that flow into files or exports."""
+
+    def test_channel_id_rejects_path_characters(self):
+        import webui_server as ws_module
+
+        assert ws_module.normalize_channel_id("@valid_name") == "valid_name"
+        for bad in ("../escape", "nested/channel", r"nested\channel", ".", "..", ""):
+            try:
+                ws_module.normalize_channel_id(bad)
+            except ValueError:
+                continue
+            raise AssertionError(f"accepted unsafe channel_id: {bad!r}")
+
+    def test_export_account_state_redacts_api_hash_by_default(self):
+        import webui_server as ws_module
+
+        state = {"label": "Work", "api_id": 123, "api_hash": "secret", "channels": {}}
+        redacted = ws_module.export_account_state(state)
+        assert redacted["api_hash"] is None
+        assert redacted["api_hash_present"] is True
+        assert state["api_hash"] == "secret"
+
+        full = ws_module.export_account_state(state, include_secrets=True)
+        assert full["api_hash"] == "secret"
+        assert "api_hash_present" not in full
+
+
 class TestContinuousOrchestrator:
     """Continuous scraping manager safety using mocked webui_server."""
 
@@ -536,27 +564,33 @@ class TestJobDeduplicationSchema:
     def test_create_job_dedup_by_account(self):
         JobRunner = self._import_job_runner()
         runner = JobRunner()
+        runner.queue.put = lambda job: None
         j1 = runner.create_job("scrape_all", "First", {"account_id": "acc1"})
         j2 = runner.create_job("scrape_all", "Second", {"account_id": "acc1"})
         assert j1.job_id == j2.job_id, "duplicate job was created instead of being reused"
         assert "Reused existing active job" in j2.logs
+        runner.shutdown(timeout=0)
 
     def test_create_job_allows_different_accounts(self):
         JobRunner = self._import_job_runner()
         runner = JobRunner()
+        runner.queue.put = lambda job: None
         j1 = runner.create_job("scrape_all", "First", {"account_id": "acc1"})
         time.sleep(0.002)  # ensure different timestamp -> different job_id
         j2 = runner.create_job("scrape_all", "Second", {"account_id": "acc2"})
         assert j1.job_id != j2.job_id
+        runner.shutdown(timeout=0)
 
     def test_create_job_allows_after_previous_completes(self):
         JobRunner = self._import_job_runner()
         runner = JobRunner()
+        runner.queue.put = lambda job: None
         j1 = runner.create_job("scrape_all", "First", {"account_id": "acc1"})
-        j1.status = "completed"
+        j1.status = "done"
         time.sleep(0.002)  # ensure different timestamp -> different job_id
         j2 = runner.create_job("scrape_all", "Second", {"account_id": "acc1"})
         assert j1.job_id != j2.job_id
+        runner.shutdown(timeout=0)
 
 
 class TestAccountHealthSummary: