Skip to content

Python API Reference

RAGuardScanner

raguard.scanner.RAGuardScanner

Main scanning engine that orchestrates detectors against RAG targets.

Source code in src/raguard/scanner.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
class RAGuardScanner:
    """Main scanning engine that orchestrates detectors against RAG targets."""

    def __init__(
        self,
        settings: RAGuardSettings | None = None,
        detectors: list[type[BaseDetector]] | None = None,
    ) -> None:
        self.settings = settings or RAGuardSettings()
        self._detectors: list[type[BaseDetector]] = detectors if detectors is not None else []
        self._load_default_detectors()

    def _load_default_detectors(self) -> None:
        """Load all registered detectors if none specified."""
        if not self._detectors:
            all_detectors = get_all_detectors()
            self._detectors = list(all_detectors.values())

    def add_detector(self, detector_cls: type[BaseDetector]) -> None:
        """Add a detector to the scanner."""
        self._detectors.append(detector_cls)

    def clear_detectors(self) -> None:
        """Clear all detectors."""
        self._detectors.clear()

    async def scan(self, target: RAGTargetConfig) -> RAGScanReport:
        """Execute a full security scan against a RAG target.

        Args:
            target: Configuration for the RAG system to test.

        Returns:
            Complete scan report with all findings.
        """
        start_time = time.monotonic()
        all_findings = []

        for detector_cls in self._detectors:
            detector = detector_cls()
            try:
                findings = await detector.detect(target)
                all_findings.extend(findings)
            except Exception as exc:
                if self.settings.debug:
                    # Log only the exception type, not the message (may contain secrets)
                    print(f"[DEBUG] Detector {detector.name} failed: {type(exc).__name__}")

        elapsed_ms = (time.monotonic() - start_time) * 1000

        report = RAGScanReport(
            target_url=target.url,
            target_type=target.type,
            findings=all_findings,
            scan_duration_ms=elapsed_ms,
            config={
                "timeout": self.settings.timeout,
                "max_retries": self.settings.max_retries,
                "embedding_model": target.embedding_model,
                "context_window": target.context_window,
            },
        )
        report.compute_risk()
        return report

    async def scan_url(
        self,
        url: str,
        target_type: TargetType = TargetType.GENERIC,
        **kwargs: str,
    ) -> RAGScanReport:
        """Convenience method to scan by URL.

        Args:
            url: Target URL.
            target_type: Type of RAG system.
            **kwargs: Additional target config options.

        Returns:
            Scan report.
        """
        target = RAGTargetConfig(
            url=url,
            type=target_type,
            **{k: v for k, v in kwargs.items() if v is not None},
        )
        return await self.scan(target)

add_detector(detector_cls)

Add a detector to the scanner.

Source code in src/raguard/scanner.py 
40
41
42
def add_detector(self, detector_cls: type[BaseDetector]) -> None:
    """Add a detector to the scanner."""
    self._detectors.append(detector_cls)

clear_detectors()

Clear all detectors.

Source code in src/raguard/scanner.py 
44
45
46
def clear_detectors(self) -> None:
    """Clear all detectors."""
    self._detectors.clear()

scan(target) async

Execute a full security scan against a RAG target.

Parameters:

Name Type Description Default
target RAGTargetConfig

Configuration for the RAG system to test.

 required

Returns:

Type Description
RAGScanReport

Complete scan report with all findings.
Source code in src/raguard/scanner.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
async def scan(self, target: RAGTargetConfig) -> RAGScanReport:
    """Execute a full security scan against a RAG target.

    Args:
        target: Configuration for the RAG system to test.

    Returns:
        Complete scan report with all findings.
    """
    start_time = time.monotonic()
    all_findings = []

    for detector_cls in self._detectors:
        detector = detector_cls()
        try:
            findings = await detector.detect(target)
            all_findings.extend(findings)
        except Exception as exc:
            if self.settings.debug:
                # Log only the exception type, not the message (may contain secrets)
                print(f"[DEBUG] Detector {detector.name} failed: {type(exc).__name__}")

    elapsed_ms = (time.monotonic() - start_time) * 1000

    report = RAGScanReport(
        target_url=target.url,
        target_type=target.type,
        findings=all_findings,
        scan_duration_ms=elapsed_ms,
        config={
            "timeout": self.settings.timeout,
            "max_retries": self.settings.max_retries,
            "embedding_model": target.embedding_model,
            "context_window": target.context_window,
        },
    )
    report.compute_risk()
    return report

scan_url(url, target_type=TargetType.GENERIC, **kwargs) async

Convenience method to scan by URL.

Parameters:

Name Type Description Default
url str

Target URL.

 required
target_type TargetType

Type of RAG system.

 GENERIC
**kwargs str

Additional target config options.

 {}

Returns:

Type Description
RAGScanReport

Scan report.
Source code in src/raguard/scanner.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
async def scan_url(
    self,
    url: str,
    target_type: TargetType = TargetType.GENERIC,
    **kwargs: str,
) -> RAGScanReport:
    """Convenience method to scan by URL.

    Args:
        url: Target URL.
        target_type: Type of RAG system.
        **kwargs: Additional target config options.

    Returns:
        Scan report.
    """
    target = RAGTargetConfig(
        url=url,
        type=target_type,
        **{k: v for k, v in kwargs.items() if v is not None},
    )
    return await self.scan(target)

Models

raguard.models.RAGTargetConfig

Bases: BaseModel

Configuration for a RAG system target.

Source code in src/raguard/models.py 
73
74
75
76
77
78
79
80
81
82
83
class RAGTargetConfig(BaseModel):
    """Configuration for a RAG system target."""

    type: TargetType = TargetType.GENERIC
    url: str = "http://localhost:8000"
    api_key: str | None = Field(default=None, exclude=True)
    collection_name: str = "default"
    embedding_model: str = "text-embedding-ada-002"
    embedding_dimension: int = 1536
    context_window: int = Field(default=4096, ge=1, le=1048576)
    metadata: dict[str, Any] = Field(default_factory=dict, description="Additional target metadata")

raguard.models.RAGFinding

Bases: BaseModel

A single security finding from a RAG scan.

Source code in src/raguard/models.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
class RAGFinding(BaseModel):
    """A single security finding from a RAG scan."""

    id: str = Field(default_factory=lambda: f"finding-{datetime.now(UTC).strftime('%Y%m%d%H%M%S%f')}")
    detector: str
    attack_type: RAGAttackType
    severity: Severity
    confidence: Confidence
    title: str = Field(..., min_length=1, max_length=500)
    description: str = Field(default="", max_length=5000)
    recommendation: str = Field(default="", max_length=2000)
    target: str = Field(default="", max_length=1024)
    risk_score: int = Field(default=0, ge=0, le=100)
    details: dict[str, Any] = Field(
        default_factory=dict,
        description="Additional finding details (not rendered in HTML reports directly)",
    )
    timestamp: str = Field(default_factory=lambda: datetime.now(UTC).isoformat())

    def model_post_init(self, __context: Any) -> None:
        if self.risk_score == 0:
            self.risk_score = self._calculate_risk()

    def _calculate_risk(self) -> int:
        base = self.severity.weight
        conf_mult = self.confidence.score
        return min(100, int(base * conf_mult * 4))

raguard.models.RAGScanReport

Bases: BaseModel

Complete scan report.

Source code in src/raguard/models.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
class RAGScanReport(BaseModel):
    """Complete scan report."""

    target_url: str
    target_type: TargetType
    findings: list[RAGFinding] = Field(default_factory=list)
    risk_score: int = 0
    risk_category: str = "none"
    scan_duration_ms: float = 0.0
    scanner_version: str = "0.1.0"
    config: dict[str, Any] = Field(default_factory=dict)
    timestamp: str = Field(default_factory=lambda: datetime.now(UTC).isoformat())

    @property
    def total_findings(self) -> int:
        return len(self.findings)

    @property
    def critical_findings(self) -> list[RAGFinding]:
        return [f for f in self.findings if f.severity == Severity.CRITICAL]

    @property
    def high_findings(self) -> list[RAGFinding]:
        return [f for f in self.findings if f.severity == Severity.HIGH]

    @property
    def summary(self) -> str:
        if not self.findings:
            return "No security issues detected."
        parts = [
            f"{len(self.critical_findings)} critical",
            f"{len(self.high_findings)} high",
            f"{len([f for f in self.findings if f.severity == Severity.MEDIUM])} medium",
            f"{len([f for f in self.findings if f.severity in (Severity.LOW, Severity.INFO)])} low/info",
        ]
        return ", ".join(parts)

    def compute_risk(self) -> None:
        """Compute aggregate risk score from findings."""
        if not self.findings:
            self.risk_score = 0
            self.risk_category = "none"
            return

        total = sum(f.risk_score for f in self.findings)
        self.risk_score = min(100, total)

        if self.risk_score >= 80:
            self.risk_category = "critical"
        elif self.risk_score >= 50:
            self.risk_category = "high"
        elif self.risk_score >= 20:
            self.risk_category = "medium"
        elif self.risk_score >= 5:
            self.risk_category = "low"
        else:
            self.risk_category = "info"

compute_risk()

Compute aggregate risk score from findings.

Source code in src/raguard/models.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def compute_risk(self) -> None:
    """Compute aggregate risk score from findings."""
    if not self.findings:
        self.risk_score = 0
        self.risk_category = "none"
        return

    total = sum(f.risk_score for f in self.findings)
    self.risk_score = min(100, total)

    if self.risk_score >= 80:
        self.risk_category = "critical"
    elif self.risk_score >= 50:
        self.risk_category = "high"
    elif self.risk_score >= 20:
        self.risk_category = "medium"
    elif self.risk_score >= 5:
        self.risk_category = "low"
    else:
        self.risk_category = "info"

raguard.models.Severity

Bases: StrEnum

Finding severity levels.

Source code in src/raguard/models.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
class Severity(StrEnum):
    """Finding severity levels."""

    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"
    INFO = "info"

    @property
    def weight(self) -> int:
        return {
            Severity.CRITICAL: 25,
            Severity.HIGH: 10,
            Severity.MEDIUM: 3,
            Severity.LOW: 1,
            Severity.INFO: 0,
        }[self]

raguard.models.Confidence

Bases: StrEnum

Detection confidence levels.

Source code in src/raguard/models.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
class Confidence(StrEnum):
    """Detection confidence levels."""

    CERTAIN = "certain"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"
    NONE = "none"

    @property
    def score(self) -> float:
        return {
            Confidence.CERTAIN: 1.0,
            Confidence.HIGH: 0.85,
            Confidence.MEDIUM: 0.6,
            Confidence.LOW: 0.3,
            Confidence.NONE: 0.0,
        }[self]

raguard.models.RAGAttackType

Bases: StrEnum

Types of RAG-specific attacks.

Source code in src/raguard/models.py 
12
13
14
15
16
17
18
19
20
21
class RAGAttackType(StrEnum):
    """Types of RAG-specific attacks."""

    DATA_POISONING = "data_poisoning"
    MEMBERSHIP_INFERENCE = "membership_inference"
    PROMPT_LEAKAGE = "prompt_leakage"
    CONTEXT_OVERFLOW = "context_overflow"
    RETRIEVAL_HIJACK = "retrieval_hijack"
    VECTOR_INJECTION = "vector_injection"
    POLICY_BYPASS = "policy_bypass"

raguard.models.TargetType

Bases: StrEnum

Supported RAG target types.

Source code in src/raguard/models.py 
64
65
66
67
68
69
70
class TargetType(StrEnum):
    """Supported RAG target types."""

    CHROMA = "chroma"
    MILVUS = "milvus"
    QDRANT = "qdrant"
    GENERIC = "generic"

Detectors

raguard.detectors.base.BaseDetector

Bases: ABC

Abstract base for all RAG security detectors.

Source code in src/raguard/detectors/base.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
class BaseDetector(ABC):
    """Abstract base for all RAG security detectors."""

    name: str = "base"
    description: str = ""

    @abstractmethod
    async def detect(self, target: RAGTargetConfig) -> list[RAGFinding]:
        """Run detection against a RAG target.

        Args:
            target: Configuration for the RAG system to test.

        Returns:
            List of security findings.
        """
        ...

detect(target) abstractmethod async

Run detection against a RAG target.

Parameters:

Name Type Description Default
target RAGTargetConfig

Configuration for the RAG system to test.

 required

Returns:

Type Description
list[RAGFinding]

List of security findings.
Source code in src/raguard/detectors/base.py 
16
17
18
19
20
21
22
23
24
25
26
@abstractmethod
async def detect(self, target: RAGTargetConfig) -> list[RAGFinding]:
    """Run detection against a RAG target.

    Args:
        target: Configuration for the RAG system to test.

    Returns:
        List of security findings.
    """
    ...

raguard.detectors.registry.get_detector(name)

Get a registered detector by name.

Source code in src/raguard/detectors/registry.py 
25
26
27
def get_detector(name: str) -> type[BaseDetector] | None:
    """Get a registered detector by name."""
    return _registry.get(name)

raguard.detectors.registry.get_all_detectors()

Get all registered detectors.

Source code in src/raguard/detectors/registry.py 
30
31
32
def get_all_detectors() -> dict[str, type[BaseDetector]]:
    """Get all registered detectors."""
    return dict(_registry)

Reporters

raguard.reporters.json.JSONReporter

JSON reporter for RAGuard scan results.

Source code in src/raguard/reporters/json.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
class JSONReporter:
    """JSON reporter for RAGuard scan results."""

    def render(self, report: RAGScanReport) -> str:
        """Render scan report as JSON.

        Args:
            report: The scan report to render.

        Returns:
            JSON string.
        """
        data = {
            "scanner": "raguard",
            "version": report.scanner_version,
            "target": {
                "url": report.target_url,
                "type": report.target_type.value,
            },
            "risk": {
                "score": report.risk_score,
                "category": report.risk_category,
            },
            "findings": [
                {
                    "id": f.id,
                    "detector": f.detector,
                    "attack_type": f.attack_type.value,
                    "severity": f.severity.value,
                    "confidence": f.confidence.value,
                    "title": f.title,
                    "description": f.description,
                    "recommendation": f.recommendation,
                    "target": f.target,
                    "risk_score": f.risk_score,
                    "details": f.details,
                    "timestamp": f.timestamp,
                }
                for f in report.findings
            ],
            "scan_duration_ms": report.scan_duration_ms,
            "timestamp": report.timestamp,
            "config": report.config,
        }
        return json.dumps(data, indent=2)

render(report)

Render scan report as JSON.

Parameters:

Name Type Description Default
report RAGScanReport

The scan report to render.

 required

Returns:

Type Description
str

JSON string.
Source code in src/raguard/reporters/json.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def render(self, report: RAGScanReport) -> str:
    """Render scan report as JSON.

    Args:
        report: The scan report to render.

    Returns:
        JSON string.
    """
    data = {
        "scanner": "raguard",
        "version": report.scanner_version,
        "target": {
            "url": report.target_url,
            "type": report.target_type.value,
        },
        "risk": {
            "score": report.risk_score,
            "category": report.risk_category,
        },
        "findings": [
            {
                "id": f.id,
                "detector": f.detector,
                "attack_type": f.attack_type.value,
                "severity": f.severity.value,
                "confidence": f.confidence.value,
                "title": f.title,
                "description": f.description,
                "recommendation": f.recommendation,
                "target": f.target,
                "risk_score": f.risk_score,
                "details": f.details,
                "timestamp": f.timestamp,
            }
            for f in report.findings
        ],
        "scan_duration_ms": report.scan_duration_ms,
        "timestamp": report.timestamp,
        "config": report.config,
    }
    return json.dumps(data, indent=2)

raguard.reporters.html.HTMLReporter

HTML reporter using Jinja2 templates.

Source code in src/raguard/reporters/html.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
class HTMLReporter:
    """HTML reporter using Jinja2 templates."""

    def render(self, report: RAGScanReport) -> str:
        """Render scan report as HTML.

        Args:
            report: The scan report to render.

        Returns:
            HTML string.
        """
        template = Template(HTML_TEMPLATE)
        return template.render(
            report=report,
            severity_critical=Severity.CRITICAL,
            severity_high=Severity.HIGH,
        )

    def render_to_file(self, report: RAGScanReport, output_path: str) -> None:
        """Render and save report to HTML file.

        Args:
            report: The scan report.
            output_path: Path to save the HTML file.
        """
        html = self.render(report)
        Path(output_path).write_text(html)

render(report)

Render scan report as HTML.

Parameters:

Name Type Description Default
report RAGScanReport

The scan report to render.

 required

Returns:

Type Description
str

HTML string.
Source code in src/raguard/reporters/html.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def render(self, report: RAGScanReport) -> str:
    """Render scan report as HTML.

    Args:
        report: The scan report to render.

    Returns:
        HTML string.
    """
    template = Template(HTML_TEMPLATE)
    return template.render(
        report=report,
        severity_critical=Severity.CRITICAL,
        severity_high=Severity.HIGH,
    )

render_to_file(report, output_path)

Render and save report to HTML file.

Parameters:

Name Type Description Default
report RAGScanReport

The scan report.

 required
output_path str

Path to save the HTML file.

 required
Source code in src/raguard/reporters/html.py 
112
113
114
115
116
117
118
119
120
def render_to_file(self, report: RAGScanReport, output_path: str) -> None:
    """Render and save report to HTML file.

    Args:
        report: The scan report.
        output_path: Path to save the HTML file.
    """
    html = self.render(report)
    Path(output_path).write_text(html)

raguard.reporters.sarif.SARIFReporter

SARIF v2.1.0 reporter for GitHub Code Scanning.

Source code in src/raguard/reporters/sarif.py 
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
class SARIFReporter:
    """SARIF v2.1.0 reporter for GitHub Code Scanning."""

    def render(self, report: RAGScanReport) -> str:
        """Render scan report as SARIF JSON.

        Args:
            report: The scan report to render.

        Returns:
            SARIF JSON string.
        """
        sarif = {
            "$schema": "https://raw.githubusercontent.com/oasis-tcs/sarif-spec/master/Schemata/sarif-schema-2.1.0.json",
            "version": "2.1.0",
            "runs": [
                {
                    "tool": {
                        "driver": {
                            "name": "RAGuard",
                            "fullName": "RAGuard RAG Security Scanner",
                            "version": report.scanner_version,
                            "informationUri": "https://github.com/Carlos-Projects/RAGuard",
                            "rules": self._build_rules(report),
                        }
                    },
                    "results": self._build_results(report),
                    "invocations": [
                        {
                            "startTimeUtc": report.timestamp,
                            "executionSuccessful": True,
                            "toolConfigurationNotifications": [],
                        }
                    ],
                }
            ],
        }
        return json.dumps(sarif, indent=2)

    def _build_rules(self, report: RAGScanReport) -> list[dict]:
        """Build SARIF rules from unique findings."""
        rules = {}
        for finding in report.findings:
            rule_id = f"RAGUARD_{finding.attack_type.value.upper()}"
            if rule_id not in rules:
                rules[rule_id] = {
                    "id": rule_id,
                    "name": finding.attack_type.value.replace("_", " ").title(),
                    "shortDescription": {"text": finding.title},
                    "fullDescription": {"text": finding.description},
                    "defaultConfiguration": {
                        "level": self._severity_to_sarif_level(finding.severity),
                    },
                    "helpUri": "https://github.com/Carlos-Projects/RAGuard",
                    "properties": {
                        "tags": ["rag-security", finding.attack_type.value],
                        "precision": "high",
                    },
                }
        return list(rules.values())

    def _build_results(self, report: RAGScanReport) -> list[dict]:
        """Build SARIF results from findings."""
        results = []
        for finding in report.findings:
            rule_id = f"RAGUARD_{finding.attack_type.value.upper()}"
            results.append(
                {
                    "ruleId": rule_id,
                    "level": self._severity_to_sarif_level(finding.severity),
                    "message": {
                        "text": finding.description,
                        "markdown": f"**{finding.title}**\n\n{finding.description}\n\n**Recommendation:** {finding.recommendation}",
                    },
                    "locations": [
                        {
                            "physicalLocation": {
                                "artifactLocation": {
                                    "uri": finding.target or report.target_url,
                                },
                            },
                        }
                    ],
                    "properties": {
                        "risk_score": finding.risk_score,
                        "confidence": finding.confidence.value,
                        "detector": finding.detector,
                    },
                }
            )
        return results

    def _severity_to_sarif_level(self, severity: Severity) -> str:
        """Map RAGuard severity to SARIF level."""
        mapping = {
            Severity.CRITICAL: "error",
            Severity.HIGH: "error",
            Severity.MEDIUM: "warning",
            Severity.LOW: "note",
            Severity.INFO: "note",
        }
        return mapping.get(severity, "note")

render(report)

Render scan report as SARIF JSON.

Parameters:

Name Type Description Default
report RAGScanReport

The scan report to render.

 required

Returns:

Type Description
str

SARIF JSON string.
Source code in src/raguard/reporters/sarif.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def render(self, report: RAGScanReport) -> str:
    """Render scan report as SARIF JSON.

    Args:
        report: The scan report to render.

    Returns:
        SARIF JSON string.
    """
    sarif = {
        "$schema": "https://raw.githubusercontent.com/oasis-tcs/sarif-spec/master/Schemata/sarif-schema-2.1.0.json",
        "version": "2.1.0",
        "runs": [
            {
                "tool": {
                    "driver": {
                        "name": "RAGuard",
                        "fullName": "RAGuard RAG Security Scanner",
                        "version": report.scanner_version,
                        "informationUri": "https://github.com/Carlos-Projects/RAGuard",
                        "rules": self._build_rules(report),
                    }
                },
                "results": self._build_results(report),
                "invocations": [
                    {
                        "startTimeUtc": report.timestamp,
                        "executionSuccessful": True,
                        "toolConfigurationNotifications": [],
                    }
                ],
            }
        ],
    }
    return json.dumps(sarif, indent=2)

raguard.reporters.console.ConsoleReporter

Rich console reporter for RAGuard scan results.

Source code in src/raguard/reporters/console.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
class ConsoleReporter:
    """Rich console reporter for RAGuard scan results."""

    def render(self, report: RAGScanReport) -> str:
        """Render scan report to Rich console output.

        Args:
            report: The scan report to render.

        Returns:
            String representation (also prints to console).
        """
        console = Console()
        output_parts = []

        color_map = {
            "none": "green",
            "low": "yellow",
            "medium": "orange1",
            "high": "red",
            "critical": "bold red",
        }
        color = color_map.get(report.risk_category, "white")

        header = Panel(
            f"[bold]Risk Score:[/] [{color}]{report.risk_score}/100 ({report.risk_category})[/]\n"
            f"[bold]Target:[/] {report.target_url} ({report.target_type.value})\n"
            f"[bold]Findings:[/] {report.total_findings} | "
            f"[bold]Duration:[/] {report.scan_duration_ms:.0f}ms",
            title="RAGuard Scan Results",
        )
        console.print(header)
        output_parts.append("RAGuard Scan Results")

        if report.summary:
            summary_panel = Panel(report.summary, title="Summary")
            console.print(summary_panel)
            output_parts.append(report.summary)

        if report.findings:
            table = Table(title=f"Findings ({len(report.findings)})")
            table.add_column("Severity", style="bold")
            table.add_column("Attack Type", style="cyan")
            table.add_column("Title")
            table.add_column("Detector")
            table.add_column("Risk")

            for f in report.findings:
                sev_style = "red" if f.severity in (Severity.CRITICAL, Severity.HIGH) else "yellow"
                table.add_row(
                    f"[{sev_style}]{f.severity.value.upper()}[/{sev_style}]",
                    f.attack_type.value,
                    f.title[:50],
                    f.detector,
                    str(f.risk_score),
                )

            console.print(table)
            output_parts.append(f"{len(report.findings)} findings")

        return "\n".join(output_parts)

render(report)

Render scan report to Rich console output.

Parameters:

Name Type Description Default
report RAGScanReport

The scan report to render.

 required

Returns:

Type Description
str

String representation (also prints to console).
Source code in src/raguard/reporters/console.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def render(self, report: RAGScanReport) -> str:
    """Render scan report to Rich console output.

    Args:
        report: The scan report to render.

    Returns:
        String representation (also prints to console).
    """
    console = Console()
    output_parts = []

    color_map = {
        "none": "green",
        "low": "yellow",
        "medium": "orange1",
        "high": "red",
        "critical": "bold red",
    }
    color = color_map.get(report.risk_category, "white")

    header = Panel(
        f"[bold]Risk Score:[/] [{color}]{report.risk_score}/100 ({report.risk_category})[/]\n"
        f"[bold]Target:[/] {report.target_url} ({report.target_type.value})\n"
        f"[bold]Findings:[/] {report.total_findings} | "
        f"[bold]Duration:[/] {report.scan_duration_ms:.0f}ms",
        title="RAGuard Scan Results",
    )
    console.print(header)
    output_parts.append("RAGuard Scan Results")

    if report.summary:
        summary_panel = Panel(report.summary, title="Summary")
        console.print(summary_panel)
        output_parts.append(report.summary)

    if report.findings:
        table = Table(title=f"Findings ({len(report.findings)})")
        table.add_column("Severity", style="bold")
        table.add_column("Attack Type", style="cyan")
        table.add_column("Title")
        table.add_column("Detector")
        table.add_column("Risk")

        for f in report.findings:
            sev_style = "red" if f.severity in (Severity.CRITICAL, Severity.HIGH) else "yellow"
            table.add_row(
                f"[{sev_style}]{f.severity.value.upper()}[/{sev_style}]",
                f.attack_type.value,
                f.title[:50],
                f.detector,
                str(f.risk_score),
            )

        console.print(table)
        output_parts.append(f"{len(report.findings)} findings")

    return "\n".join(output_parts)

Utilities

raguard.utils.http.RAGuardHTTPClient

Async HTTP client with retries and timeouts for RAG scanning.

Security features: - SSRF protection via URL validation - Redirect following disabled by default - Configurable timeouts and retries

Source code in src/raguard/utils/http.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
class RAGuardHTTPClient:
    """Async HTTP client with retries and timeouts for RAG scanning.

    Security features:
    - SSRF protection via URL validation
    - Redirect following disabled by default
    - Configurable timeouts and retries
    """

    def __init__(
        self,
        timeout: float = 30.0,
        max_retries: int = 3,
        headers: dict[str, str] | None = None,
        follow_redirects: bool = False,
    ) -> None:
        self.timeout = timeout
        self.max_retries = max_retries
        self.headers = headers or {}
        self.follow_redirects = follow_redirects
        self._client: httpx.AsyncClient | None = None

    async def get_client(self) -> httpx.AsyncClient:
        """Get or create the HTTP client."""
        if self._client is None:
            self._client = httpx.AsyncClient(
                timeout=self.timeout,
                headers=self.headers,
                follow_redirects=self.follow_redirects,
            )
        return self._client

    async def close(self) -> None:
        """Close the HTTP client."""
        if self._client:
            await self._client.aclose()
            self._client = None

    async def get(self, url: str, **kwargs: Any) -> httpx.Response:
        """Perform GET request with retries and SSRF validation."""
        validate_target_url(url)
        client = await self.get_client()
        last_exc: Exception | None = None

        for attempt in range(self.max_retries):
            try:
                response = await client.get(url, **kwargs)
                return response
            except httpx.RequestError as exc:
                last_exc = exc
                if attempt < self.max_retries - 1:
                    continue
                raise

        raise last_exc or httpx.RequestError("Max retries exceeded")

    async def post(self, url: str, json: dict | None = None, **kwargs: Any) -> httpx.Response:
        """Perform POST request with retries and SSRF validation."""
        validate_target_url(url)
        client = await self.get_client()
        last_exc: Exception | None = None

        for attempt in range(self.max_retries):
            try:
                response = await client.post(url, json=json, **kwargs)
                return response
            except httpx.RequestError as exc:
                last_exc = exc
                if attempt < self.max_retries - 1:
                    continue
                raise

        raise last_exc or httpx.RequestError("Max retries exceeded")

    async def __aenter__(self) -> RAGuardHTTPClient:
        return self

    async def __aexit__(self, *args: Any) -> None:
        await self.close()

close() async

Close the HTTP client.

Source code in src/raguard/utils/http.py 
104
105
106
107
108
async def close(self) -> None:
    """Close the HTTP client."""
    if self._client:
        await self._client.aclose()
        self._client = None

get(url, **kwargs) async

Perform GET request with retries and SSRF validation.

Source code in src/raguard/utils/http.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
async def get(self, url: str, **kwargs: Any) -> httpx.Response:
    """Perform GET request with retries and SSRF validation."""
    validate_target_url(url)
    client = await self.get_client()
    last_exc: Exception | None = None

    for attempt in range(self.max_retries):
        try:
            response = await client.get(url, **kwargs)
            return response
        except httpx.RequestError as exc:
            last_exc = exc
            if attempt < self.max_retries - 1:
                continue
            raise

    raise last_exc or httpx.RequestError("Max retries exceeded")

get_client() async

Get or create the HTTP client.

Source code in src/raguard/utils/http.py 
 94
 95
 96
 97
 98
 99
100
101
102
async def get_client(self) -> httpx.AsyncClient:
    """Get or create the HTTP client."""
    if self._client is None:
        self._client = httpx.AsyncClient(
            timeout=self.timeout,
            headers=self.headers,
            follow_redirects=self.follow_redirects,
        )
    return self._client

post(url, json=None, **kwargs) async

Perform POST request with retries and SSRF validation.

Source code in src/raguard/utils/http.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
async def post(self, url: str, json: dict | None = None, **kwargs: Any) -> httpx.Response:
    """Perform POST request with retries and SSRF validation."""
    validate_target_url(url)
    client = await self.get_client()
    last_exc: Exception | None = None

    for attempt in range(self.max_retries):
        try:
            response = await client.post(url, json=json, **kwargs)
            return response
        except httpx.RequestError as exc:
            last_exc = exc
            if attempt < self.max_retries - 1:
                continue
            raise

    raise last_exc or httpx.RequestError("Max retries exceeded")

raguard.utils.embeddings.cosine_similarity(a, b)

Compute cosine similarity between two vectors.

Parameters:

Name Type Description Default
a ndarray

First vector.

 required
b ndarray

Second vector.

 required

Returns:

Type Description
float

Cosine similarity score between -1 and 1.
Source code in src/raguard/utils/embeddings.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
    """Compute cosine similarity between two vectors.

    Args:
        a: First vector.
        b: Second vector.

    Returns:
        Cosine similarity score between -1 and 1.
    """
    a = np.asarray(a, dtype=np.float64)
    b = np.asarray(b, dtype=np.float64)

    if a.ndim == 1:
        a = a.reshape(1, -1)
    if b.ndim == 1:
        b = b.reshape(1, -1)

    dot_product = np.dot(a, b.T)
    norm_a = np.linalg.norm(a, axis=1, keepdims=True)
    norm_b = np.linalg.norm(b, axis=1, keepdims=True)

    if norm_a.item() == 0 or norm_b.item() == 0:
        return 0.0

    return float((dot_product / (norm_a * norm_b.T)).flatten()[0])