fix(hosted): prevent credential disclosure by gating mentions and redacting published results

deploy/coven-github/coven_github_adapter.py

@@ -290,6 +290,14 @@ def trigger_enabled(policy, trigger):
     return trigger in set(enabled or [])
 
 
+def comment_author_is_trusted(comment):
+    return str(comment.get("author_association") or "").upper() in {
+        "OWNER",
+        "MEMBER",
+        "COLLABORATOR",
+    }
+
+
 def build_task_from_event(event_name, delivery_id, payload, policy):
     repository = payload.get("repository") or {}
     installation = payload.get("installation") or {}
@@ -319,6 +327,8 @@ def build_task_from_event(event_name, delivery_id, payload, policy):
             return ignored(base, "issue_comment_without_mention")
         if not trigger_enabled(policy, "issue_mention"):
             return ignored(base, "issue_mention_not_enabled")
+        if not comment_author_is_trusted(comment):
+            return ignored(base, "issue_mention_author_not_trusted")
         if issue.get("pull_request"):
             base.update(
                 {
@@ -356,6 +366,8 @@ def build_task_from_event(event_name, delivery_id, payload, policy):
             return ignored(base, "pr_review_comment_without_mention")
         if not trigger_enabled(policy, "pr_review_comment"):
             return ignored(base, "pr_review_comment_not_enabled")
+        if not comment_author_is_trusted(comment):
+            return ignored(base, "pr_review_comment_author_not_trusted")
         base.update(
             {
                 "trigger": "pr_review_comment",
@@ -1249,7 +1261,10 @@ def publish_result_if_configured(task, result_path, token):
         task["publication_state"] = "publication_skipped_no_issue_or_pr_number"
         return
 
-    body = publication_comment_body(task, result)
+    body = redact_secrets(
+        publication_comment_body(task, result),
+        [token, load_codex_access_token()],
+    )
     repo = task.get("repository")
     url = "https://api.github.com/repos/{}/issues/{}/comments".format(repo, int(number))
     try:
@@ -1649,6 +1664,19 @@ def redact_tokenish(text):
     return redacted
 
 
+def redact_secrets(text, secret_values=()):
+    redacted = str(text)
+    for secret in secret_values:
+        if secret:
+            redacted = redacted.replace(str(secret), "[redacted]")
+    redacted = redact_tokenish(redacted)
+    return re.sub(
+        r"(?<![A-Za-z0-9])sk-[A-Za-z0-9_-]{12,}",
+        "sk-[redacted]",
+        redacted,
+    )
+
+
 def fail_task(path, task, reason, detail):
     task["state"] = "failed"
     task["failure_category"] = reason

deploy/coven-github/test_coven_github_adapter.py

@@ -13,6 +13,84 @@ def load_adapter():
 
 
 class HostedAdapterTests(unittest.TestCase):
+    def test_untrusted_comment_authors_cannot_queue_tasks(self):
+        adapter = load_adapter()
+        policy = {"bot_usernames": ["cody"]}
+        common = {
+            "repository": {"id": 1, "full_name": "OpenCoven/coven-github"},
+            "installation": {"id": 2},
+        }
+
+        issue_task = adapter.build_task_from_event(
+            "issue_comment",
+            "issue-delivery",
+            {
+                **common,
+                "issue": {"number": 3},
+                "comment": {"body": "@cody fix this", "author_association": "NONE"},
+            },
+            policy,
+        )
+        review_task = adapter.build_task_from_event(
+            "pull_request_review_comment",
+            "review-delivery",
+            {
+                **common,
+                "pull_request": {"number": 4},
+                "comment": {"body": "@cody fix this", "author_association": "CONTRIBUTOR"},
+            },
+            policy,
+        )
+
+        self.assertEqual(issue_task["state"], "ignored")
+        self.assertEqual(issue_task["ignored_reason"], "issue_mention_author_not_trusted")
+        self.assertEqual(review_task["state"], "ignored")
+        self.assertEqual(review_task["ignored_reason"], "pr_review_comment_author_not_trusted")
+
+    def test_trusted_comment_author_can_queue_task(self):
+        adapter = load_adapter()
+        task = adapter.build_task_from_event(
+            "issue_comment",
+            "trusted-delivery",
+            {
+                "repository": {"id": 1, "full_name": "OpenCoven/coven-github"},
+                "installation": {"id": 2},
+                "issue": {"number": 3},
+                "comment": {"body": "@cody fix this", "author_association": "MEMBER"},
+            },
+            {"bot_usernames": ["cody"]},
+        )
+
+        self.assertEqual(task["state"], "queued")
+
+    def test_published_comment_redacts_runtime_credentials(self):
+        adapter = load_adapter()
+        published = {}
+        adapter.load_codex_access_token = lambda: "sk-runtime-secret-1234567890"
+
+        def fake_github_request(method, url, token, body=None):
+            published.update(body)
+            return {"id": 1, "html_url": "https://example.test/comment/1"}
+
+        adapter.github_request = fake_github_request
+        with tempfile.TemporaryDirectory() as tmp:
+            result_path = Path(tmp) / "result.json"
+            result_path.write_text(
+                '{"status":"success","summary":"ghs_installation-secret and '
+                'sk-runtime-secret-1234567890"}',
+                encoding="utf-8",
+            )
+            task = {
+                "publication": {"mode": "comment"},
+                "repository": "OpenCoven/coven-github",
+                "task": {"issue_number": 3},
+            }
+            adapter.publish_result_if_configured(task, result_path, "ghs_installation-secret")
+
+        self.assertNotIn("installation-secret", published["body"])
+        self.assertNotIn("runtime-secret", published["body"])
+        self.assertIn("[redacted]", published["body"])
+
     def test_mentions_are_boundary_aware(self):
         adapter = load_adapter()
         policy = {"bot_usernames": ["cody"]}