fix: Address review - path traversal, guardrail fallback, elif chain, retries, smoke tests

Author: sudsali

scripts/issue_bot/bedrock_client.py

@@ -17,10 +17,11 @@ def __init__(self, cfg):
             config=BotoConfig(
                 read_timeout=cfg.bedrock_timeout,
                 connect_timeout=cfg.bedrock_timeout,
-                retries={"max_attempts": 1, "mode": "standard"},
+                retries={"max_attempts": 3, "mode": "adaptive"},
             ),
         )
         self._guardrail_id = cfg.guardrail_id
+        self._guardrail_version = cfg.guardrail_version
         self._failures = 0
         self._circuit_open = False
 
@@ -66,7 +67,7 @@ def invoke(self, prompt, max_tokens=4096, temperature=0.3, json_schema=None):
             if self._guardrail_id:
                 kwargs["guardrailConfig"] = {
                     "guardrailIdentifier": self._guardrail_id,
-                    "guardrailVersion": "1",
+                    "guardrailVersion": self._guardrail_version,
                     "trace": "enabled",
                 }
 
@@ -103,7 +104,9 @@ def _split_prompt(self, prompt):
         marker = "<knowledge_base>"
         idx = prompt.find(marker)
         if idx == -1:
-            # No marker — send everything as user content (unguarded)
+            # No marker — guard everything as untrusted
+            if self._guardrail_id:
+                return None, [{"guardContent": {"text": {"text": prompt}}}]
             return None, [{"text": prompt}]
 
         system = prompt[:idx].strip()

scripts/issue_bot/config.py

@@ -22,6 +22,7 @@ def __init__(self):
 
         self.slack_webhook_url = os.getenv("SLACK_WEBHOOK_URL", "")
         self.guardrail_id = os.getenv("GUARDRAIL_ID", "")
+        self.guardrail_version = os.getenv("GUARDRAIL_VERSION", "DRAFT")
 
         self.dry_run = os.getenv("DRY_RUN", "false").lower() == "true"
         self.enable_slack = bool(self.slack_webhook_url)

scripts/issue_bot/github_client.py

@@ -90,10 +90,11 @@ def get_codebase_map(self, src_dir="pydeequ"):
             return ""
 
     def read_local_file(self, path):
-        if not path.startswith("pydeequ/") and not path.startswith("tests/") and not path.startswith("tutorials/"):
-            logger.error(f"Blocked read outside allowed dirs: {path}")
+        full_path = os.path.realpath(os.path.join(self._repo_root, path))
+        repo_root = os.path.realpath(self._repo_root)
+        if not full_path.startswith(repo_root + os.sep):
+            logger.error(f"Blocked path traversal: {path}")
             return ""
-        full_path = os.path.join(self._repo_root, path)
         try:
             with open(full_path, "r", errors="replace") as f:
                 return f.read()
@@ -185,9 +186,10 @@ def _get_valid_diff_lines(self, number):
                     continue
                 if line.startswith("-"):
                     continue
-                if line.startswith("+") or not line.startswith("\\"):
-                    lines.add(current_line)
-                    current_line += 1
+                if line.startswith("\\"):
+                    continue
+                lines.add(current_line)
+                current_line += 1
             valid[path] = lines
         return valid
 

scripts/issue_bot/main.py

@@ -10,7 +10,6 @@
 import os
 import datetime
 import logging
-from string import Template
 
 from .config import Config
 from .bedrock_client import BedrockClient
@@ -27,10 +26,12 @@
 
 
 def _render(template_str, **kwargs):
-    """Render a prompt template safely. Converts {var} to $var for Template.safe_substitute
-    so untrusted content like PR bodies containing {braces} won't crash or leak."""
-    converted = template_str.replace("{", "${")
-    return Template(converted).safe_substitute(**kwargs)
+    """Render a prompt template safely using str.replace per known variable.
+    Unlike .format(), this won't crash on untrusted content containing {braces}."""
+    result = template_str
+    for key, value in kwargs.items():
+        result = result.replace("{" + key + "}", str(value))
+    return result
 
 
 def _load_schema(name):
@@ -275,7 +276,7 @@ def act():
                 slack.send_escalation(number, title, html_url, labels)
             logger.info(f"Responded to #{number}")
 
-    if action == "ESCALATE":
+    elif action == "ESCALATE":
         reason = result.get("reason", "")
         if reason == "user_dissatisfied":
             ack = (
@@ -309,7 +310,7 @@ def act():
         slack.send_escalation(number, title, html_url, labels)
         logger.info(f"Escalated #{number}")
 
-    if action == "CLOSE" and not is_pr:
+    elif action == "CLOSE" and not is_pr:
         msg = (
             "This issue may not be related to the PyDeequ data quality library. "
             "The maintainer team has been notified and will review." + footer

tests/test_bot.py

@@ -215,3 +215,25 @@ def test_strips_preamble(self):
     def test_preserves_normal_text(self):
         text = "This is a normal response."
         assert _clean_response(text) == text
+
+
+# ---------------------------------------------------------------------------
+# Smoke test — module imports without crashing
+# ---------------------------------------------------------------------------
+
+class TestSmoke:
+    def test_main_module_imports(self):
+        from issue_bot import main
+        assert hasattr(main, 'analyze')
+        assert hasattr(main, 'act')
+
+    def test_sanitizer_imports(self):
+        from issue_bot import sanitizer
+        assert hasattr(sanitizer, 'sanitize')
+
+    def test_schemas_loadable(self):
+        from issue_bot.main import ISSUE_RESPONSE_SCHEMA, PR_REVIEW_SCHEMA, FOLLOWUP_SCHEMA
+        import json
+        assert json.loads(ISSUE_RESPONSE_SCHEMA)["type"] == "object"
+        assert json.loads(PR_REVIEW_SCHEMA)["type"] == "object"
+        assert json.loads(FOLLOWUP_SCHEMA)["type"] == "object"