Analyzer Pipeline: Writing Custom Detection Rules

Analyzers are where detection happens. Each analyzer encodes a single heuristic or pattern, queries the PostgreSQL tables populated by collectors, and emits structured findings when suspicious activity is found. GitHub Threat Detector ships with 17 built-in rules spread across seven analyzer modules. Every rule — built-in or custom — extends the same BaseAnalyzer abstract class, so the interface is consistent and new detectors slot in with minimal boilerplate.

`BaseAnalyzer` Interface

All detection rules inherit from analyzers/base.py. The class is abstract: you must define rule_id, severity, and implement run(). The emit() method is provided by the base class and handles the database write for you.

from abc import ABC, abstractmethod
from db.queries import insert_finding


class BaseAnalyzer(ABC):
    rule_id: str
    severity: str

    def emit(
        self,
        repo_name: str,
        description: str,
        actor_login: str | None = None,
        event_id: str | None = None,
        evidence: dict | None = None,
    ) -> None:
        insert_finding(
            rule_id=self.rule_id,
            severity=self.severity,
            repo_name=repo_name,
            description=description,
            actor_login=actor_login,
            event_id=event_id,
            evidence=evidence,
        )

    @abstractmethod
    def run(self, repo_name: str | None = None) -> int:
        ...

Class attributes

rule_id

str (required)

A unique, machine-readable identifier for this detection rule, e.g. "workflow-file-change" or "pr-target-abuse". This value is stored in findings.rule_id and used by the --rules CLI flag to select individual analyzers.

severity

str (required)

Risk level of findings emitted by this analyzer. Must be one of "critical", "high", "medium", or "low" — enforced by the CHECK constraint on findings.severity.

`emit()` — writing a finding

Call self.emit() once for each suspicious match your run() method identifies. The method delegates to insert_finding() in db/queries.py, which executes an INSERT … ON CONFLICT DO NOTHING so duplicate findings are silently dropped.

repo_name

str (required)

The affected repository in owner/repo format.

description

str (required)

A human-readable sentence describing the threat, shown in the report output. Include the actor login and any relevant context so the finding is self-explanatory without needing to look up the evidence blob.

GitHub username of the actor associated with the suspicious activity. Pass None for repo-level findings with no identifiable actor.

event_id

str | None

The github_events.id of the event that triggered this finding. When set, the deduplication index idx_findings_dedup_event on (rule_id, event_id) prevents re-emitting the same finding across runs. Pass None for findings derived from workflow_runs or static file analysis.

evidence

dict | None

A free-form dictionary of supporting data serialised to JSONB. Common patterns include commit SHAs, PR numbers, workflow run IDs, and timestamp strings. There is no enforced schema — choose keys that make triage easier.

`run()` — the detection logic

@abstractmethod
def run(self, repo_name: str | None = None) -> int:
    ...

run() receives an optional repo_name. When called with None, the analyzer should scan all repositories in the database. When called with a specific owner/repo string, it should restrict its query to that repository. Return the number of findings emitted. The CLI uses this count to print a summary line after each analyzer completes.

Built-in Analyzers

The seven built-in modules and their exported ANALYZERS lists are concatenated by _load_analyzers() in cli.py:

def _load_analyzers() -> list:
    from analyzers.malicious_ci import ANALYZERS as a1
    from analyzers.release_tamper import ANALYZERS as a2
    from analyzers.social_eng import ANALYZERS as a3
    from analyzers.fork_bomb import ANALYZERS as a4
    from analyzers.push_anomaly import ANALYZERS as a5
    from analyzers.ai_workflow_abuse import ANALYZERS as a6
    from analyzers.commit_tamper import ANALYZERS as a7
    return a1 + a2 + a3 + a4 + a5 + a6 + a7

Example: `analyzers/malicious_ci.py`

The malicious CI module ships three rules that illustrate the full range of the detection model:

class WorkflowFileChangeAnalyzer(BaseAnalyzer):
    """PushEvent touching .github/workflows/ by an actor who is not a known maintainer."""

    rule_id = "workflow-file-change"
    severity = "high"

    def run(self, repo_name: str | None = None) -> int:
        sql = """
            WITH maintainers AS (
                SELECT DISTINCT actor_login
                FROM github_events
                WHERE event_type = 'PushEvent'
                  AND created_at < now() - INTERVAL '30 days'
                  AND (%(repo)s IS NULL OR repo_name = %(repo)s)
            ),
            workflow_pushes AS (
                SELECT id, repo_name, actor_login, created_at, payload
                FROM github_events
                WHERE event_type = 'PushEvent'
                  AND payload::text ILIKE '%/.github/workflows/%'
                  AND (%(repo)s IS NULL OR repo_name = %(repo)s)
            )
            SELECT wp.*
            FROM workflow_pushes wp
            LEFT JOIN maintainers m ON wp.actor_login = m.actor_login
            WHERE m.actor_login IS NULL
        """
        count = 0
        with get_cursor(dict_cursor=True) as cur:
            cur.execute(sql, {"repo": repo_name})
            for row in cur.fetchall():
                self.emit(
                    repo_name=row["repo_name"],
                    description=f"Actor {row['actor_login']} pushed workflow file changes but has no prior commit history",
                    actor_login=row["actor_login"],
                    event_id=row["id"],
                    evidence={"created_at": str(row["created_at"])},
                )
                count += 1
        return count

Writing a Custom Analyzer

To add a new detection rule, create a new Python module in the analyzers/ directory, subclass BaseAnalyzer, and export an ANALYZERS list. The steps are always the same:

Set rule_id — a unique kebab-case string.
Set severity — one of critical, high, medium, low.
Implement run(self, repo_name) — query the database, call self.emit() for each match, return the count.
Export ANALYZERS — a list of instantiated analyzer objects at module level.
Register the module — add an import to _load_analyzers() in cli.py.

Use the (%(repo)s IS NULL OR repo_name = %(repo)s) pattern in every WHERE clause. This single conditional makes your analyzer work correctly for both global runs (python cli.py analyze) and scoped runs (python cli.py analyze --repos owner/repo) without duplicating the query.

Complete example: `LargeCommitPushAnalyzer`

from db.client import get_cursor
from analyzers.base import BaseAnalyzer


class LargeCommitPushAnalyzer(BaseAnalyzer):
    rule_id = "large-commit-push"
    severity = "medium"

    def run(self, repo_name: str | None = None) -> int:
        sql = """
            SELECT id, repo_name, actor_login, created_at,
                   jsonb_array_length(payload->'commits') AS commit_count
            FROM github_events
            WHERE event_type = 'PushEvent'
              AND jsonb_array_length(payload->'commits') > 50
              AND (%(repo)s IS NULL OR repo_name = %(repo)s)
        """
        count = 0
        with get_cursor(dict_cursor=True) as cur:
            cur.execute(sql, {"repo": repo_name})
            for row in cur.fetchall():
                self.emit(
                    repo_name=row["repo_name"],
                    description=f"Actor {row['actor_login']} pushed {row['commit_count']} commits at once",
                    actor_login=row["actor_login"],
                    event_id=row["id"],
                    evidence={"commit_count": row["commit_count"]},
                )
                count += 1
        return count


ANALYZERS = [LargeCommitPushAnalyzer()]

Registering a Custom Analyzer

Once your module is written, add it to _load_analyzers() in cli.py:

def _load_analyzers() -> list:
    from analyzers.malicious_ci import ANALYZERS as a1
    from analyzers.release_tamper import ANALYZERS as a2
    from analyzers.social_eng import ANALYZERS as a3
    from analyzers.fork_bomb import ANALYZERS as a4
    from analyzers.push_anomaly import ANALYZERS as a5
    from analyzers.ai_workflow_abuse import ANALYZERS as a6
    from analyzers.commit_tamper import ANALYZERS as a7
    from analyzers.push_volume import ANALYZERS as a8   # ← your new module
    return a1 + a2 + a3 + a4 + a5 + a6 + a7 + a8

After registering, verify the new rule appears in a dry run:

python cli.py analyze --rules large-commit-push

Analyzer Execution Model

When python cli.py analyze is called, the CLI:

Calls _load_analyzers() to build the full list of analyzer instances.
Optionally filters the list by --rules (comma-separated rule_id values).
Determines the repo list from --repos or DEFAULT_REPOS; falls back to [None] (scan all repos).
Iterates over every (analyzer, repo) pair and calls analyzer.run(repo_name=repo).
Prints a per-rule summary and a grand total of findings emitted.

Exceptions raised inside run() are caught and printed as warnings — one failing analyzer does not abort the rest of the pipeline.

Running workflow-file-change on acme-corp/payments-service...
  → 2 finding(s)
Running pr-target-abuse on acme-corp/payments-service...
  → clean
Running large-commit-push on acme-corp/payments-service...
  → 1 finding(s)

Analysis complete. Total findings: 3

Get Started

CLI Reference

Threat Detection

Architecture

Case Studies

Analyzer Pipeline: Writing Custom Detection Rules

`BaseAnalyzer` Interface

Class attributes

`emit()` — writing a finding

`run()` — the detection logic

Built-in Analyzers

Example: `analyzers/malicious_ci.py`

Writing a Custom Analyzer

Complete example: `LargeCommitPushAnalyzer`

Registering a Custom Analyzer

Analyzer Execution Model

Build docs developers (and LLMs) love

Get Started

CLI Reference

Threat Detection

Architecture

Case Studies

Documentation Index

​BaseAnalyzer Interface

​Class attributes

​emit() — writing a finding

​run() — the detection logic

​Built-in Analyzers

​Example: analyzers/malicious_ci.py

​Writing a Custom Analyzer

​Complete example: LargeCommitPushAnalyzer

​Registering a Custom Analyzer

​Analyzer Execution Model

Build docs developers (and LLMs) love

`BaseAnalyzer` Interface

Class attributes

`emit()` — writing a finding

`run()` — the detection logic

Built-in Analyzers

Example: `analyzers/malicious_ci.py`

Writing a Custom Analyzer

Complete example: `LargeCommitPushAnalyzer`

Registering a Custom Analyzer

Analyzer Execution Model