Fixed ruff types

This commit is contained in:
AlphaRex-pixel 2026-08-15 16:25:59 +05:30
parent 648780c40a
commit d8442a7e05
4 changed files with 141 additions and 144 deletions

View file

@ -61,14 +61,18 @@ class TealEngine:
self._pii_policy = next((p for p in policies if p["type"] == "pii"), None)
self._cost_policy = next((p for p in policies if p["type"] == "cost"), None)
self._tool_policy = next((p for p in policies if p["type"] == "tool_auth"), None)
self._tool_policy = next(
(p for p in policies if p["type"] == "tool_auth"), None
)
def evaluate_text(self, text: str) -> Decision:
correlation_id: Final = str(uuid.uuid4())
if not text or not self._pii_policy:
return Decision(Action.ALLOW.value, "NO_VIOLATIONS", correlation_id)
findings: Final = tuple(name for name, pat in self.patterns.items() if pat.search(text))
findings: Final = tuple(
name for name, pat in self.patterns.items() if pat.search(text)
)
if not findings:
return Decision(Action.ALLOW.value, "NO_VIOLATIONS", correlation_id)
@ -107,20 +111,27 @@ class TealEngine:
def check_tool(self, tool_name: str) -> bool:
if not self._tool_policy:
return True
allowlist: Final = self._tool_policy.get("allowlist")
blocklist: Final = self._tool_policy.get("blocklist")
if blocklist and tool_name in blocklist:
raw_blocklist: Final = self._tool_policy.get("blocklist")
blocklist: Final = (
raw_blocklist if isinstance(raw_blocklist, (list, tuple)) else ()
)
if tool_name in blocklist:
return False
if allowlist is not None:
raw_allowlist: Final = self._tool_policy.get("allowlist")
if raw_allowlist is not None:
allowlist: Final = (
raw_allowlist if isinstance(raw_allowlist, (list, tuple)) else ()
)
return tool_name in allowlist
return True
def check_budget(self, session_id: str = "default") -> tuple[bool, float, float]:
if not self._cost_policy:
return False, 0.0, 0.0
limit: Final = self._cost_policy.get("daily_limit_usd")
if limit is None:
raw_limit: Final = self._cost_policy.get("daily_limit_usd")
if not isinstance(raw_limit, (int, float)):
return False, 0.0, 0.0
limit: Final[float] = float(raw_limit)
today: Final = date.today().isoformat() # noqa: DTZ011 # daily budget boundary is intentionally local-server-time, not UTC
with self._lock:
spent: Final = self._spend_by_day.get(today, 0.0)

View file

@ -1,7 +1,7 @@
"""
Regex-based PII / secrets detection patterns for TealTiger.
47 patterns across five categories: government IDs, financial data,
40+ patterns across five categories: government IDs, financial data,
contact info, network/device identifiers, and credentials/API keys.
These are deterministic regex matches — fast (no ML, no network calls) but
@ -10,135 +10,95 @@ the loosest patterns (e.g. generic credit card, generic driver's license).
Callers can narrow scope via TealEngine(policies=[{"patterns": [...]}])
instead of "all", or override this module's PII_PATTERNS entirely.
"""
from __future__ import annotations
import re
from collections.abc import Mapping
from types import MappingProxyType
from typing import Final
PII_PATTERNS: Final = MappingProxyType(
{
# ---------- government / national IDs ----------
"ssn": re.compile(r"\b\d{3}-\d{2}-\d{4}\b"),
"ssn_no_dash": re.compile(r"\b(?!000|666|9\d{2})\d{3}(?!00)\d{2}(?!0000)\d{4}\b"),
"itin": re.compile(r"\b9\d{2}-(7\d|8[0-8])-\d{4}\b"),
"ein": re.compile(r"\b\d{2}-\d{7}\b"),
"us_passport": re.compile(r"\b[A-Z]{1,2}\d{7,9}\b"),
"passport_generic_intl": re.compile(r"\b[A-Z]{2}\d{6,9}\b"),
"drivers_license_generic": re.compile(r"\b[A-Z]{1,2}\d{6,8}\b"),
"medicare_id_us": re.compile(r"\b\d[A-Z]\d{2}-[A-Z]\d{2}-[A-Z]\d{2}\b"),
"npi_number": re.compile(r"\b\d{10}\b"),
"uk_nino": re.compile(r"\b[A-CEGHJ-PR-TW-Z]{2}\d{6}[A-D]\b"),
"date_of_birth": re.compile(r"\b(0[1-9]|1[0-2])[/-](0[1-9]|[12]\d|3[01])[/-](19|20)\d{2}\b"),
"vin": re.compile(r"\b[A-HJ-NPR-Z0-9]{17}\b"),
# ---------- financial ----------
"credit_card_visa": re.compile(r"\b4\d{3}[ -]?\d{4}[ -]?\d{4}[ -]?\d{4}\b"),
"credit_card_mastercard": re.compile(r"\b5[1-5]\d{2}[ -]?\d{4}[ -]?\d{4}[ -]?\d{4}\b"),
"credit_card_amex": re.compile(r"\b3[47]\d{2}[ -]?\d{6}[ -]?\d{5}\b"),
"credit_card_discover": re.compile(r"\b6(?:011|5\d{2})[ -]?\d{4}[ -]?\d{4}[ -]?\d{4}\b"),
"credit_card_generic": re.compile(r"\b(?:\d[ -]*?){13,16}\b"),
"iban": re.compile(r"\b[A-Z]{2}\d{2}[A-Z0-9]{10,30}\b"),
"swift_bic": re.compile(r"\b[A-Z]{4}[A-Z]{2}[A-Z0-9]{2}(?:[A-Z0-9]{3})?\b"),
"us_bank_routing": re.compile(r"\b\d{9}\b"),
"us_bank_account": re.compile(r"\b\d{8,17}\b"),
"bitcoin_address": re.compile(r"\b(bc1|[13])[a-zA-HJ-NP-Z0-9]{25,39}\b"),
"ethereum_address": re.compile(r"\b0x[a-fA-F0-9]{40}\b"),
# ---------- contact info ----------
"email": re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b"),
"phone_us": re.compile(r"\b(?:\+?1[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b"),
"phone_intl": re.compile(r"\+\d{1,3}[-.\s]?\(?\d{1,4}\)?(?:[-.\s]?\d{2,4}){2,4}\b"),
"us_zip_plus4": re.compile(r"\b\d{5}-\d{4}\b"),
"po_box": re.compile(r"\bP\.?O\.?\s?Box\s?\d+\b", re.IGNORECASE),
"lat_long_coordinates": re.compile(r"\b-?\d{1,3}\.\d{3,},\s?-?\d{1,3}\.\d{3,}\b"),
# ---------- network / device identifiers ----------
"ipv4": re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b"),
"ipv6": re.compile(r"\b(?:[A-Fa-f0-9]{1,4}:){7}[A-Fa-f0-9]{1,4}\b"),
"mac_address": re.compile(r"\b(?:[0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}\b"),
"imei": re.compile(r"\b\d{15}\b"),
"url_with_credentials": re.compile(r"\bhttps?://[^:\s]+:[^@\s]+@[^\s]+\b"),
# ---------- credentials / API keys / secrets ----------
"aws_access_key": re.compile(r"\b(AKIA|ASIA)[0-9A-Z]{16}\b"),
"aws_secret_key": re.compile(r"\b(?=.*[A-Za-z])(?=.*\d)(?=.*[+/])[A-Za-z0-9+/]{40}\b"),
"gcp_api_key": re.compile(r"\bAIza[0-9A-Za-z_-]{35}\b"),
"azure_client_secret": re.compile(r"\b[A-Za-z0-9_~.-]{3}\dQ~[A-Za-z0-9_~.-]{31,34}\b"),
"github_token": re.compile(r"\bgh[pousr]_[A-Za-z0-9]{36,255}\b"),
"gitlab_token": re.compile(r"\bglpat-[A-Za-z0-9_-]{20}\b"),
"slack_token": re.compile(r"\bxox[baprs]-[A-Za-z0-9-]{10,72}\b"),
"stripe_live_key": re.compile(r"\b(sk|pk)_live_[A-Za-z0-9]{24,}\b"),
"openai_api_key": re.compile(r"\bsk-[A-Za-z0-9]{20,}\b"),
"jwt": re.compile(r"\beyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\b"),
"private_key_pem": re.compile(r"-----BEGIN (RSA |EC |OPENSSH |)PRIVATE KEY-----"),
"generic_bearer_token": re.compile(r"\bBearer\s+[A-Za-z0-9_\-.]{20,}\b"),
"basic_auth_header": re.compile(r"\bBasic\s+[A-Za-z0-9+/]{20,}={0,2}\b"),
}
)
PII_PATTERNS = {
# ---------- government / national IDs ----------
"ssn": re.compile(r"\b\d{3}-\d{2}-\d{4}\b"),
"ssn_no_dash": re.compile(r"\b(?!000|666|9\d{2})\d{3}(?!00)\d{2}(?!0000)\d{4}\b"),
"itin": re.compile(r"\b9\d{2}-(7\d|8[0-8])-\d{4}\b"),
"ein": re.compile(r"\b\d{2}-\d{7}\b"),
"us_passport": re.compile(r"\b[A-Z]{1,2}\d{7,9}\b"),
"passport_generic_intl": re.compile(r"\b[A-Z]{2}\d{6,9}\b"),
"drivers_license_generic": re.compile(r"\b[A-Z]{1,2}\d{6,8}\b"),
"medicare_id_us": re.compile(r"\b\d[A-Z]\d{2}-[A-Z]\d{2}-[A-Z]\d{2}\b"),
"npi_number": re.compile(r"\b\d{10}\b"),
"uk_nino": re.compile(r"\b[A-CEGHJ-PR-TW-Z]{2}\d{6}[A-D]\b"),
"date_of_birth": re.compile(r"\b(0[1-9]|1[0-2])[/-](0[1-9]|[12]\d|3[01])[/-](19|20)\d{2}\b"),
"vin": re.compile(r"\b[A-HJ-NPR-Z0-9]{17}\b"),
# ---------- financial ----------
"credit_card_visa": re.compile(r"\b4\d{3}[ -]?\d{4}[ -]?\d{4}[ -]?\d{4}\b"),
"credit_card_mastercard": re.compile(r"\b5[1-5]\d{2}[ -]?\d{4}[ -]?\d{4}[ -]?\d{4}\b"),
"credit_card_amex": re.compile(r"\b3[47]\d{2}[ -]?\d{6}[ -]?\d{5}\b"),
"credit_card_discover": re.compile(r"\b6(?:011|5\d{2})[ -]?\d{4}[ -]?\d{4}[ -]?\d{4}\b"),
"credit_card_generic": re.compile(r"\b(?:\d[ -]*?){13,16}\b"),
"iban": re.compile(r"\b[A-Z]{2}\d{2}[A-Z0-9]{10,30}\b"),
"swift_bic": re.compile(r"\b[A-Z]{4}[A-Z]{2}[A-Z0-9]{2}(?:[A-Z0-9]{3})?\b"),
"us_bank_routing": re.compile(r"\b\d{9}\b"),
"us_bank_account": re.compile(r"\b\d{8,17}\b"),
"bitcoin_address": re.compile(r"\b(bc1|[13])[a-zA-HJ-NP-Z0-9]{25,39}\b"),
"ethereum_address": re.compile(r"\b0x[a-fA-F0-9]{40}\b"),
# ---------- contact info ----------
"email": re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b"),
"phone_us": re.compile(r"\b(?:\+?1[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b"),
"phone_intl": re.compile(r"\+\d{1,3}[-.\s]?\(?\d{1,4}\)?(?:[-.\s]?\d{2,4}){2,4}\b"),
"us_zip_plus4": re.compile(r"\b\d{5}-\d{4}\b"),
"po_box": re.compile(r"\bP\.?O\.?\s?Box\s?\d+\b", re.IGNORECASE),
"lat_long_coordinates": re.compile(r"\b-?\d{1,3}\.\d{3,},\s?-?\d{1,3}\.\d{3,}\b"),
# ---------- network / device identifiers ----------
"ipv4": re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b"),
"ipv6": re.compile(r"\b(?:[A-Fa-f0-9]{1,4}:){7}[A-Fa-f0-9]{1,4}\b"),
"mac_address": re.compile(r"\b(?:[0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}\b"),
"imei": re.compile(r"\b\d{15}\b"),
"url_with_credentials": re.compile(r"\bhttps?://[^:\s]+:[^@\s]+@[^\s]+\b"),
# ---------- credentials / API keys / secrets ----------
"aws_access_key": re.compile(r"\b(AKIA|ASIA)[0-9A-Z]{16}\b"),
"aws_secret_key": re.compile(r"\b(?=.*[A-Za-z])(?=.*\d)(?=.*[+/])[A-Za-z0-9+/]{40}\b"),
"gcp_api_key": re.compile(r"\bAIza[0-9A-Za-z_-]{35}\b"),
"azure_client_secret": re.compile(r"\b[A-Za-z0-9_~.-]{3}\dQ~[A-Za-z0-9_~.-]{31,34}\b"),
"github_token": re.compile(r"\bgh[pousr]_[A-Za-z0-9]{36,255}\b"),
"gitlab_token": re.compile(r"\bglpat-[A-Za-z0-9_-]{20}\b"),
"slack_token": re.compile(r"\bxox[baprs]-[A-Za-z0-9-]{10,72}\b"),
"stripe_live_key": re.compile(r"\b(sk|pk)_live_[A-Za-z0-9]{24,}\b"),
"openai_api_key": re.compile(r"\bsk-[A-Za-z0-9]{20,}\b"),
"jwt": re.compile(r"\beyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\b"),
"private_key_pem": re.compile(r"-----BEGIN (RSA |EC |OPENSSH |)PRIVATE KEY-----"),
"generic_bearer_token": re.compile(r"\bBearer\s+[A-Za-z0-9_\-.]{20,}\b"),
"basic_auth_header": re.compile(r"\bBasic\s+[A-Za-z0-9+/]{20,}={0,2}\b"),
}
def default_patterns() -> Mapping[str, re.Pattern[str]]:
"""Return the built-in, immutable pattern mapping."""
return PII_PATTERNS
def default_patterns() -> dict:
"""Return a fresh copy of the built-in pattern set."""
return dict(PII_PATTERNS)
PATTERN_CATEGORIES: Final = MappingProxyType(
{
"government_id": (
"ssn",
"ssn_no_dash",
"itin",
"ein",
"us_passport",
"passport_generic_intl",
"drivers_license_generic",
"medicare_id_us",
"npi_number",
"uk_nino",
"date_of_birth",
"vin",
),
"financial": (
"credit_card_visa",
"credit_card_mastercard",
"credit_card_amex",
"credit_card_discover",
"credit_card_generic",
"iban",
"swift_bic",
"us_bank_routing",
"us_bank_account",
"bitcoin_address",
"ethereum_address",
),
"contact": (
"email",
"phone_us",
"phone_intl",
"us_zip_plus4",
"po_box",
"lat_long_coordinates",
),
"network_device": (
"ipv4",
"ipv6",
"mac_address",
"imei",
"url_with_credentials",
),
"credentials": (
"aws_access_key",
"aws_secret_key",
"gcp_api_key",
"azure_client_secret",
"github_token",
"gitlab_token",
"slack_token",
"stripe_live_key",
"openai_api_key",
"jwt",
"private_key_pem",
"generic_bearer_token",
"basic_auth_header",
),
}
)
PATTERN_CATEGORIES = {
"government_id": [
"ssn", "ssn_no_dash", "itin", "ein", "us_passport", "passport_generic_intl",
"drivers_license_generic", "medicare_id_us", "npi_number", "uk_nino",
"date_of_birth", "vin",
],
"financial": [
"credit_card_visa", "credit_card_mastercard", "credit_card_amex",
"credit_card_discover", "credit_card_generic", "iban", "swift_bic",
"us_bank_routing", "us_bank_account", "bitcoin_address", "ethereum_address",
],
"contact": [
"email", "phone_us", "phone_intl", "us_zip_plus4", "po_box",
"lat_long_coordinates",
],
"network_device": [
"ipv4", "ipv6", "mac_address", "imei", "url_with_credentials",
],
"credentials": [
"aws_access_key", "aws_secret_key", "gcp_api_key", "azure_client_secret",
"github_token", "gitlab_token", "slack_token", "stripe_live_key",
"openai_api_key", "jwt", "private_key_pem", "generic_bearer_token",
"basic_auth_header",
],
}

View file

@ -32,7 +32,9 @@ _EMPTY_MAPPING: Final = MappingProxyType({})
DEFAULT_POLICIES: Final = (
MappingProxyType({"type": "pii", "action": "REDACT", "patterns": "all"}),
MappingProxyType({"type": "cost", "action": "ENFORCE", "daily_limit_usd": 50.0}),
MappingProxyType({"type": "tool_auth", "action": "ENFORCE", "allowlist": None}), # None = allow all
MappingProxyType(
{"type": "tool_auth", "action": "ENFORCE", "allowlist": None}
), # None = allow all
)
@ -41,7 +43,7 @@ class TealTigerGuardrail(CustomGuardrail):
self,
policies: Sequence[Mapping[str, object]] | None = None,
policy_mode: str = "ENFORCE",
**kwargs: object, # kwargs-ok: forwards unknown/future CustomGuardrail.__init__ params (guardrail_name, event_hook, default_on, ...) to the base class
**kwargs, # kwargs-ok: forwards unknown/future CustomGuardrail.__init__ params (guardrail_name, event_hook, default_on, ...) to the base class
) -> None:
self.engine: Final = TealEngine(
policies=policies or DEFAULT_POLICIES,
@ -71,19 +73,29 @@ class TealTigerGuardrail(CustomGuardrail):
self._check_tool_calls(inputs)
self._check_budget(request_data)
checked_texts: Final = tuple(self._check_text_or_raise(text) for text in inputs.get("texts") or ())
checked_texts: Final = tuple(
self._check_text_or_raise(text) for text in inputs.get("texts") or ()
)
return {**inputs, "texts": list(checked_texts)} # mutable-ok: dict interop
def _check_tool_calls(self, inputs: GenericGuardrailAPIInputs) -> None:
for tool_call in inputs.get("tool_calls") or ():
tool_name = tool_call.get("name") or tool_call.get("function", _EMPTY_MAPPING).get("name")
tool_name = tool_call.get("name") or tool_call.get(
"function", _EMPTY_MAPPING
).get("name")
if tool_name and not self.engine.check_tool(tool_name):
raise ValueError(f"TealTiger: blocked — TOOL_NOT_ALLOWLISTED ({tool_name})")
raise ValueError(
f"TealTiger: blocked — TOOL_NOT_ALLOWLISTED ({tool_name})"
)
def _check_budget(self, request_data: Mapping[str, object]) -> None:
over_budget, spent, limit = self.engine.check_budget(session_id=request_data.get("user") or "default")
raw_user: Final = request_data.get("user")
session_id: Final[str] = raw_user if isinstance(raw_user, str) else "default"
over_budget, spent, limit = self.engine.check_budget(session_id=session_id)
if over_budget:
raise ValueError(f"TealTiger: blocked — DAILY_BUDGET_EXCEEDED (${spent:.2f} / ${limit:.2f})")
raise ValueError(
f"TealTiger: blocked — DAILY_BUDGET_EXCEEDED (${spent:.2f} / ${limit:.2f})"
)
def _check_text_or_raise(self, text: str) -> str:
decision: Final = self.engine.evaluate_text(text)

View file

@ -66,6 +66,7 @@ from collections import Counter
from collections.abc import Callable, Iterator, Mapping, Sequence
from pathlib import Path
from typing import Final, NamedTuple
import sys
REPO_ROOT = Path(__file__).resolve().parent.parent
BUDGET_PATH = REPO_ROOT / "basedpyright-code-budget.json"
@ -172,11 +173,18 @@ def typecheck_env_commands(env_dir: Path = TYPECHECK_ENV_DIR) -> tuple[tuple[str
*(("--python", python_pin) if python_pin else ()),
*(flag for group in TYPECHECK_DEP_GROUPS for flag in ("--group", group)),
)
generate: Final = (str(env_dir / "bin" / "python"), str(PRISMA_GENERATE_SCRIPT))
if sys.platform == "win32":
python_executable = env_dir / "Scripts" / "python.exe"
else:
python_executable = env_dir / "bin" / "python"
generate: Final = (str(python_executable), str(PRISMA_GENERATE_SCRIPT))
return (sync, generate)
def _run_provision_step(cmd: tuple[str, ...], env: Mapping[str, str]) -> int:
print("TYPECHECK DEBUG CMD:", list(cmd))
print("TYPECHECK DEBUG CWD:", REPO_ROOT)
proc = subprocess.run(
list(cmd), cwd=REPO_ROOT, env=dict(env), capture_output=True, text=True
)
@ -220,7 +228,13 @@ def run_basedpyright(cwd: Path = REPO_ROOT, env_dir: Path = TYPECHECK_ENV_DIR) -
diagnostics) whenever the repo has one. Exit 0 (clean) and 1 (errors
found) are both output-bearing runs; anything else is a crash and fails
loudly instead of reading as zero errors."""
bin_dir: Final = env_dir / "bin"
if sys.platform == "win32":
bin_dir = env_dir / "Scripts"
python_executable = bin_dir / "python.exe"
else:
bin_dir = env_dir / "bin"
python_executable = bin_dir / "python"
str(python_executable)
proc = subprocess.run(
[
str(bin_dir / "basedpyright"),