Skip to content

Custom Patterns

TornadoAPI Guard allows you to add custom patterns for detecting suspicious activity.


Adding Custom Patterns

Add your own patterns to the detection system:

from tornadoapi_guard import sus_patterns_handler

async def setup_patterns():
    # Add custom pattern
    await sus_patterns_handler.add_pattern(
        r"malicious_pattern.*",
        custom=True
    )

Pattern Types

You can add patterns for different types of attacks:

# Custom XSS pattern
await sus_patterns_handler.add_pattern(
    r"<script\s*src=.*>",
    custom=True
)

# Custom SQL injection pattern
await sus_patterns_handler.add_pattern(
    r";\s*DROP\s+TABLE",
    custom=True
)

# Custom file path pattern
await sus_patterns_handler.add_pattern(
    r"\.\.\/.*\/etc\/passwd",
    custom=True
)

Managing Patterns

Remove or modify existing patterns:

# Remove a custom pattern
success = await sus_patterns_handler.remove_pattern(
    r"malicious_pattern.*",
    custom=True
)
if success:
    print("Pattern removed successfully")
else:
    print("Pattern not found")

# Get all patterns (both default and custom)
all_patterns = await sus_patterns_handler.get_all_patterns()

# Get only default patterns
default_patterns = await sus_patterns_handler.get_default_patterns()

# Get only custom patterns
custom_patterns = await sus_patterns_handler.get_custom_patterns()

# Get all compiled patterns
all_compiled_patterns = await sus_patterns_handler.get_all_compiled_patterns()

# Get only default compiled patterns
default_compiled = await sus_patterns_handler.get_default_compiled_patterns()

# Get only custom compiled patterns
custom_compiled = await sus_patterns_handler.get_custom_compiled_patterns()

Pattern Testing

Test your patterns against requests:

from guard_core.utils import detect_penetration_attempt
from tornadoapi_guard import SecurityHandler
from tornadoapi_guard.adapters import TornadoGuardRequest


class TestPatternsHandler(SecurityHandler):
    async def post(self) -> None:
        guard_request = TornadoGuardRequest(self)
        is_suspicious, trigger_info = await detect_penetration_attempt(guard_request)
        self.write({
            "suspicious": is_suspicious,
            "trigger_info": trigger_info,
            "request_body": self.request.body.decode("utf-8", errors="replace"),
        })