How to Detect AI-Generated Code in PRs with Codeium 1.8 and SonarQube 10.5 for Python 3.13 Projects

\nimport os\nimport sys\nimport json\nimport logging\nimport requests\nfrom typing import Dict, List, Optional\nfrom dataclasses import dataclass\n\n# Configure logging for debug output\nlogging.basicConfig(\n    level=logging.INFO,\n    format=\"%(asctime)s - %(levelname)s - %(message)s\"\n)\nlogger = logging.getLogger(__name__)\n\n@dataclass\nclass CodeiumConfig:\n    \"\"\"Configuration for Codeium 1.8 self-hosted gateway\"\"\"\n    gateway_url: str\n    api_key: str\n    confidence_threshold: float = 0.85  # Minimum confidence to flag AI code\n    python_version: str = \"3.13\"\n\nclass CodeiumClient:\n    \"\"\"Client for Codeium 1.8 AI fingerprinting API\"\"\"\n\n    def __init__(self, config: CodeiumConfig):\n        self.config = config\n        self.session = requests.Session()\n        self.session.headers.update({\n            \"Authorization\": f\"Bearer {config.api_key}\",\n            \"Content-Type\": \"application/json\",\n            \"X-Codeium-Version\": \"1.8\"\n        })\n\n    def scan_pr_diff(self, pr_diff: str, repo_name: str) -> Optional[Dict]:\n        \"\"\"\n        Scan a PR diff for AI-generated code using Codeium 1.8 fingerprinting.\n        \n        Args:\n            pr_diff: Raw PR diff string from GitHub/GitLab API\n            repo_name: Full repo name (e.g., \"owner/repo\") for context\n            \n        Returns:\n            Dictionary of scan results, or None if scan fails\n        \"\"\"\n        endpoint = f\"{self.config.gateway_url}/v1/scan/diff\"\n        payload = {\n            \"diff\": pr_diff,\n            \"repo\": repo_name,\n            \"language\": \"python\",\n            \"language_version\": self.config.python_version,\n            \"threshold\": self.config.confidence_threshold\n        }\n\n        try:\n            response = self.session.post(endpoint, json=payload, timeout=10)\n            response.raise_for_status()  # Raise HTTPError for 4xx/5xx responses\n            return response.json()\n        except requests.exceptions.Timeout:\n            logger.error(\"Codeium API request timed out after 10 seconds\")\n        except requests.exceptions.HTTPError as e:\n            logger.error(f\"Codeium API returned error: {e.response.status_code} - {e.response.text}\")\n        except json.JSONDecodeError:\n            logger.error(\"Failed to parse Codeium API response as JSON\")\n        except Exception as e:\n            logger.error(f\"Unexpected error scanning PR diff: {str(e)}\")\n        return None\n\n    def generate_report(self, scan_results: Dict) -> str:\n        \"\"\"Generate human-readable report from scan results\"\"\"\n        if not scan_results:\n            return \"No scan results available\"\n        \n        report_lines = [\n            f\"Codeium 1.8 Scan Report for {scan_results.get('repo', 'unknown repo')}\",\n            f\"Total snippets scanned: {scan_results.get('total_snippets', 0)}\",\n            f\"AI-flagged snippets: {scan_results.get('flagged_count', 0)}\",\n            f\"Average confidence: {scan_results.get('avg_confidence', 0.0):.2f}\",\n            \"\\nFlagged Snippets:\"\n        ]\n\n        for snippet in scan_results.get(\"flagged_snippets\", []):\n            report_lines.append(\n                f\"  - Line {snippet['start_line']}-{snippet['end_line']}: \"\n                f\"Confidence {snippet['confidence']:.2f}, Model: {snippet.get('model', 'unknown')}\"\n            )\n\n        return \"\\n\".join(report_lines)\n\ndef main():\n    # Load config from environment variables (best practice for CI)\n    gateway_url = os.getenv(\"CODEIUM_GATEWAY_URL\")\n    api_key = os.getenv(\"CODEIUM_API_KEY\")\n    repo_name = os.getenv(\"GITHUB_REPOSITORY\")  # Set automatically in GitHub Actions\n\n    if not all([gateway_url, api_key, repo_name]):\n        logger.error(\"Missing required environment variables: CODEIUM_GATEWAY_URL, CODEIUM_API_KEY, GITHUB_REPOSITORY\")\n        sys.exit(1)\n\n    config = CodeiumConfig(gateway_url=gateway_url, api_key=api_key)\n    client = CodeiumClient(config)\n\n    # Sample PR diff for testing (Python 3.13 code with match-case)\n    sample_diff = \"\"\"diff --git a/main.py b/main.py\nindex 1234567..abcdefg 100644\n--- a/main.py\n+++ b/main.py\n@@ -1,5 +1,12 @@\n+from typing import TypeVar, Generic\n+\n+T = TypeVar('T')\n+\n+class Response(Generic[T]):\n+    def __init__(self, data: T, status: int):\n+        self.data = data\n+        self.status = status\n+\n def handle_request(req: dict):\n-    if req.get(\"type\") == \"user\":\n-        return {\"status\": 200, \"data\": req[\"user\"]}\n-    elif req.get(\"type\") == \"admin\":\n-        return {\"status\": 200, \"data\": req[\"admin\"]}\n+    match req.get(\"type\"):\n+        case \"user\":\n+            return Response(data=req[\"user\"], status=200)\n+        case \"admin\":\n+            return Response(data=req[\"admin\"], status=200)\n+        case _:\n+            return Response(data=None, status=404)\n\"\"\"\n\n    logger.info(f\"Scanning sample PR diff for repo {repo_name}\")\n    results = client.scan_pr_diff(sample_diff, repo_name)\n\n    if results:\n        print(client.generate_report(results))\n    else:\n        logger.error(\"Scan failed, no results to report\")\n        sys.exit(1)\n\nif __name__ == \"__main__\":\n    main()\n

\npackage org.sonar.python.rules.ai;\n\nimport java.util.List;\nimport java.util.ArrayList;\nimport org.sonar.api.rule.RuleKey;\nimport org.sonar.api.rule.RuleScope;\nimport org.sonar.api.rule.RuleStatus;\nimport org.sonar.api.rules.RuleType;\nimport org.sonar.api.server.rule.RulesDefinition;\nimport org.sonar.check.Rule;\nimport org.sonar.check.Priority;\nimport org.sonar.python.checks.AbstractCheck;\nimport org.sonar.python.tree.PythonTree;\nimport org.sonar.python.tree.FunctionDefTree;\nimport org.sonar.python.tree.ClassDefTree;\nimport org.sonar.python.tree.ExpressionTree;\nimport org.sonar.python.tree.MatchStatementTree;\nimport org.sonar.python.tree.TypeAnnotationTree;\n\n/**\n * SonarQube 10.5 custom rule to flag AI-generated Python 3.13 code with common anti-patterns.\n * Targets patterns prevalent in GPT-4o, Claude 3.5, and Gemini 1.5 generated code.\n */\n@Rule(\n    key = \"AI001\",\n    name = \"AI-Generated Code Anti-Pattern Detection\",\n    description = \"Flags Python 3.13 code snippets with high confidence of being AI-generated based on common anti-patterns: missing error handling in match-case, unused generic type parameters, redundant type annotations.\",\n    priority = Priority.MAJOR,\n    type = RuleType.CODE_SMELL,\n    scope = RuleScope.ALL,\n    status = RuleStatus.READY\n)\npublic class AiCodeDetectionRule extends AbstractCheck {\n\n    private static final String RULE_KEY = \"AI001\";\n    private static final double CONFIDENCE_THRESHOLD = 0.8;\n\n    private final List flaggedPatterns = new ArrayList<>();\n\n    @Override\n    public void visitMatchStatement(MatchStatementTree matchStatement) {\n        // AI-generated match-case often omits error handling for unmatched cases\n        if (matchStatement.cases().stream().noneMatch(c -> c.pattern() instanceof org.sonar.python.tree.WildcardPatternTree)) {\n            flaggedPatterns.add(String.format(\n                \"Match statement at line %d missing wildcard case (common AI anti-pattern)\",\n                matchStatement.firstToken().line()\n            ));\n        }\n        super.visitMatchStatement(matchStatement);\n    }\n\n    @Override\n    public void visitClassDefinition(ClassDefTree classDef) {\n        // AI-generated generic classes often declare unused TypeVar parameters\n        if (classDef.typeParameters() != null) {\n            classDef.typeParameters().typeParameters().forEach(typeParam -> {\n                boolean isUsed = classDef.descendants().stream()\n                    .anyMatch(node -> node instanceof TypeAnnotationTree && \n                        ((TypeAnnotationTree) node).type().toString().contains(typeParam.name().name()));\n                if (!isUsed) {\n                    flaggedPatterns.add(String.format(\n                        \"Unused TypeVar %s in class %s at line %d (common AI anti-pattern)\",\n                        typeParam.name().name(), classDef.name().name(), typeParam.firstToken().line()\n                    ));\n                }\n            });\n        }\n        super.visitClassDefinition(classDef);\n    }\n\n    @Override\n    public void visitFunctionDefinition(FunctionDefTree functionDef) {\n        // AI-generated functions often have redundant type annotations for trivial returns\n        if (functionDef.returnTypeAnnotation() != null) {\n            String returnType = functionDef.returnTypeAnnotation().type().toString();\n            if (returnType.equals(\"None\") && functionDef.body().statements().size() == 1) {\n                flaggedPatterns.add(String.format(\n                    \"Redundant None return type annotation in function %s at line %d (common AI anti-pattern)\",\n                    functionDef.name().name(), functionDef.firstToken().line()\n                ));\n            }\n        }\n        super.visitFunctionDefinition(functionDef);\n    }\n\n    @Override\n    public void leaveFile(PythonTree tree) {\n        // Report all flagged patterns as SonarQube issues\n        flaggedPatterns.forEach(pattern -> {\n            addIssue(tree.firstToken(), RULE_KEY, pattern)\n                .withSeverity(\"MAJOR\")\n                .withConfidence(CONFIDENCE_THRESHOLD);\n        });\n        flaggedPatterns.clear(); // Clear for next file\n    }\n\n    /**\n     * Define rule metadata for SonarQube 10.5 rule repository\n     */\n    public static class AiRuleDefinition implements RulesDefinition.NewRepository {\n        @Override\n        public RulesDefinition.NewRepository setKey(String s) {\n            return this;\n        }\n\n        @Override\n        public RulesDefinition.NewRule createRule(String ruleKey) {\n            RulesDefinition.NewRule rule = new RulesDefinition.NewRule() {\n                @Override\n                public RulesDefinition.NewRule setName(String s) {\n                    return this;\n                }\n\n                @Override\n                public RulesDefinition.NewRule setMarkdownDescription(String s) {\n                    return this;\n                }\n\n                @Override\n                public RulesDefinition.NewRule setType(RuleType ruleType) {\n                    return this;\n                }\n\n                @Override\n                public RulesDefinition.NewRule setSeverity(String s) {\n                    return this;\n                }\n\n                @Override\n                public RulesDefinition.NewRule addTags(String... strings) {\n                    return this;\n                }\n            };\n            return rule;\n        }\n\n        @Override\n        public RulesDefinition.NewRule createRule(String s, String s1) {\n            return null;\n        }\n\n        @Override\n        public RulesDefinition.NewRepository done() {\n            return this;\n        }\n    }\n}\n

\nimport os\nimport sys\nimport json\nimport logging\nimport requests\nfrom typing import Dict, List, Tuple\nfrom dataclasses import dataclass\n\n# Configure logging\nlogging.basicConfig(\n    level=logging.INFO,\n    format=\"%(asctime)s - %(levelname)s - %(message)s\"\n)\nlogger = logging.getLogger(__name__)\n\n@dataclass\nclass ScanResult:\n    source: str  # \"codeium\" or \"sonarqube\"\n    flagged: bool\n    confidence: float\n    line_ranges: List[Tuple[int, int]]\n    description: str\n\nclass IntegratedAIDetector:\n    \"\"\"Combines Codeium 1.8 and SonarQube 10.5 results for unified PR reporting\"\"\"\n\n    def __init__(self, codeium_client, sonarqube_client):\n        self.codeium = codeium_client\n        self.sonarqube = sonarqube_client\n\n    def run_combined_scan(self, pr_diff: str, repo_name: str, pr_number: int) -> List[ScanResult]:\n        \"\"\"\n        Run both Codeium and SonarQube scans, combine results.\n        \n        Args:\n            pr_diff: Raw PR diff string\n            repo_name: Full repo name (owner/repo)\n            pr_number: GitHub PR number for reporting\n            \n        Returns:\n            List of combined ScanResult objects\n        \"\"\"\n        combined_results = []\n\n        # Run Codeium scan\n        codeium_results = self.codeium.scan_pr_diff(pr_diff, repo_name)\n        if codeium_results:\n            for snippet in codeium_results.get(\"flagged_snippets\", []):\n                combined_results.append(ScanResult(\n                    source=\"codeium\",\n                    flagged=True,\n                    confidence=snippet[\"confidence\"],\n                    line_ranges=[(snippet[\"start_line\"], snippet[\"end_line\"])],\n                    description=f\"Codeium flagged: {snippet.get('model', 'unknown model')} generated code\"\n                ))\n\n        # Run SonarQube scan (assumes SonarQube has already analyzed the PR branch)\n        sonar_results = self.sonarqube.get_pr_issues(repo_name, pr_number)\n        if sonar_results:\n            for issue in sonar_results.get(\"issues\", []):\n                if issue[\"rule\"].startswith(\"AI\"):  # Only AI-specific rules\n                    combined_results.append(ScanResult(\n                        source=\"sonarqube\",\n                        flagged=True,\n                        confidence=issue.get(\"confidence\", 0.0),\n                        line_ranges=[(issue[\"line\"], issue[\"line\"])],\n                        description=f\"SonarQube flagged: {issue['message']}\"\n                    ))\n\n        return combined_results\n\n    def generate_pr_comment(self, results: List[ScanResult], pr_number: int) -> str:\n        \"\"\"Generate GitHub PR comment with combined results\"\"\"\n        if not results:\n            return f\"## AI Code Detection Results for PR #{pr_number}\\n\\n✅ No AI-generated code detected.\"\n\n        comment_lines = [\n            f\"## AI Code Detection Results for PR #{pr_number}\",\n            f\"⚠️ {len(results)} AI-suspect snippet(s) detected via Codeium 1.8 and SonarQube 10.5\",\n            \"\\n### Flagged Snippets:\"\n        ]\n\n        for idx, result in enumerate(results, 1):\n            comment_lines.append(f\"\\n#### {idx}. {result.source.upper()} Flag (Confidence: {result.confidence:.2f})\")\n            comment_lines.append(f\"**Lines**: {result.line_ranges[0][0]}-{result.line_ranges[0][1]}\")\n            comment_lines.append(f\"**Description**: {result.description}\")\n            comment_lines.append(f\"**Remediation**: Review code for AI-specific anti-patterns, add unit tests, verify logic manually.\")\n\n        comment_lines.append(\"\\n---\\n*This comment was generated automatically by the AI Code Detection Pipeline*\")\n        return \"\\n\".join(comment_lines)\n\n    def post_pr_comment(self, repo_name: str, pr_number: int, comment: str) -> bool:\n        \"\"\"Post comment to GitHub PR using GitHub API\"\"\"\n        github_token = os.getenv(\"GITHUB_TOKEN\")\n        if not github_token:\n            logger.error(\"Missing GITHUB_TOKEN environment variable\")\n            return False\n\n        url = f\"https://api.github.com/repos/{repo_name}/issues/{pr_number}/comments\"\n        headers = {\n            \"Authorization\": f\"Bearer {github_token}\",\n            \"Accept\": \"application/vnd.github.v3+json\"\n        }\n        payload = {\"body\": comment}\n\n        try:\n            response = requests.post(url, headers=headers, json=payload, timeout=10)\n            response.raise_for_status()\n            logger.info(f\"Successfully posted comment to PR #{pr_number}\")\n            return True\n        except requests.exceptions.HTTPError as e:\n            logger.error(f\"Failed to post PR comment: {e.response.status_code} - {e.response.text}\")\n        except Exception as e:\n            logger.error(f\"Unexpected error posting PR comment: {str(e)}\")\n        return False\n\ndef main():\n    # Load environment variables\n    repo_name = os.getenv(\"GITHUB_REPOSITORY\")\n    pr_number = os.getenv(\"PR_NUMBER\")\n    pr_diff = os.getenv(\"PR_DIFF\")  # Assumes diff is passed as env var or fetched via API\n\n    if not all([repo_name, pr_number, pr_diff]):\n        logger.error(\"Missing required environment variables: GITHUB_REPOSITORY, PR_NUMBER, PR_DIFF\")\n        sys.exit(1)\n\n    # Initialize clients (reuse CodeiumClient from Step 1, assume SonarQubeClient exists)\n    codeium_config = CodeiumConfig(\n        gateway_url=os.getenv(\"CODEIUM_GATEWAY_URL\"),\n        api_key=os.getenv(\"CODEIUM_API_KEY\")\n    )\n    codeium_client = CodeiumClient(codeium_config)\n    sonarqube_client = SonarQubeClient(  # Assume this is defined elsewhere\n        server_url=os.getenv(\"SONARQUBE_URL\"),\n        token=os.getenv(\"SONARQUBE_TOKEN\")\n    )\n\n    # Run combined scan\n    detector = IntegratedAIDetector(codeium_client, sonarqube_client)\n    results = detector.run_combined_scan(pr_diff, repo_name, int(pr_number))\n\n    # Generate and post comment\n    comment = detector.generate_pr_comment(results, int(pr_number))\n    success = detector.post_pr_comment(repo_name, int(pr_number), comment)\n\n    sys.exit(0 if success else 1)\n\nif __name__ == \"__main__\":\n    main()\n

\n{\n  \"python\": {\n    \"3.13\": {\n      \"confidence_threshold\": 0.78,\n      \"exclude_patterns\": [\"**/migrations/**\", \"**/generated/**\", \"**/proto/**\"],\n      \"model_whitelist\": [\"gpt-4o\", \"claude-3.5-sonnet\", \"gemini-1.5-pro\"]\n    }\n  }\n}\n  

\n@Override\npublic void visitClassDefinition(ClassDefTree classDef) {\n    if (classDef.typeParameters() != null) {\n        classDef.typeParameters().typeParameters().forEach(typeParam -> {\n            boolean isUsed = classDef.descendants().stream()\n                .anyMatch(node -> node instanceof TypeAnnotationTree && \n                    ((TypeAnnotationTree) node).type().toString().contains(typeParam.name().name()));\n            if (!isUsed) {\n                addIssue(typeParam, \"AI002\", String.format(\n                    \"Unused TypeVar %s in generic class (common AI anti-pattern)\",\n                    typeParam.name().name()\n                ));\n            }\n        });\n    }\n    super.visitClassDefinition(classDef);\n}\n  

\n- name: Post AI Detection Results to PR\n  uses: actions/github-script@v7\n  with:\n    script: |\n      const comment = `## AI Code Detection Results\n      ⚠️ 2 AI-suspect snippets detected\n      ### Snippet 1: Lines 12-18 (Codeium Confidence: 0.92)\n      **Model**: GPT-4o\n      **Anti-Pattern**: Missing wildcard case in match statement\n      **Remediation**:\n      1. Add a \`case _:\` clause to handle unmatched types\n      2. Write a unit test for unmatched input\n      3. Verify return type matches Response generic parameter\n      [View Full Report]({{ codeium_report_url }})`;\n      github.rest.issues.createComment({\n        owner: context.repo.owner,\n        repo: context.repo.repo,\n        issue_number: context.issue.number,\n        body: comment\n      });\n