Security Decorators

The decorators module provides route-level security controls that can be applied to individual Tornado handler methods. These decorators offer fine-grained control over security policies on a per-route basis, complementing the global SecurityMiddleware security features.

The decorator implementation lives in guard_core and is re-exported through tornadoapi_guard. The API surface is identical to fastapi-guard — only the application pattern differs, since Tornado dispatches to handler methods instead of plain route functions.

---

Overview

Security decorators allow you to:

• Apply specific security rules to individual routes
• Override global security settings for specific endpoints
• Combine multiple security measures in a clean, readable way
• Implement behavioral analysis and monitoring per endpoint

---

Setup

Create the SecurityDecorator once at module level and register it with the middleware before any handler is dispatched:

from tornadoapi_guard import (
    SecurityConfig,
    SecurityDecorator,
    SecurityHandler,
    SecurityMiddleware,
)

config = SecurityConfig()
middleware = SecurityMiddleware(config=config)
guard_deco = SecurityDecorator(config)
middleware.set_decorator_handler(guard_deco)

Pass both the middleware and the decorator as Application settings so the framework and the pipeline can resolve them:

app = tornado.web.Application(
    routes,
    security_middleware=middleware,
    guard_decorator=guard_deco,
)

Decorators must be applied to handler methods (get, post, put, ...), never to the handler class itself.

---

Main Decorator Class

SecurityDecorator

guard_core.decorators.SecurityDecorator(config)

Bases: BaseSecurityDecorator, AccessControlMixin, RateLimitingMixin, BehavioralMixin, AuthenticationMixin, ContentFilteringMixin, AdvancedMixin

Source code in guard_core/decorators/base.py

def __init__(self, config: SecurityConfig) -> None:
    from guard_core.handlers.behavior_handler import BehaviorTracker

    self.config = config
    self._route_configs: dict[str, RouteConfig] = {}
    # id -> id(func) of the function the id was derived from. A stamped
    # _guard_route_id is only reused when THIS instance assigned it to
    # THIS function; otherwise a cross-instance collision would select
    # another route's config.
    self._route_id_owner: dict[str, int] = {}
    self._route_config_revision = RouteConfigRevision()
    self.behavior_tracker = BehaviorTracker(config)
    self.agent_handler: Any = None
    self.geo_ip_handler: Any = None

The main decorator class that combines all security capabilities. This is the primary class you'll use in your application.

Example Usage:

class SensitiveHandler(SecurityHandler):
    @guard_deco.rate_limit(requests=5, window=300)
    @guard_deco.require_ip(whitelist=["10.0.0.0/8"])
    @guard_deco.block_countries(["CN", "RU"])
    async def get(self) -> None:
        self.write({"data": "sensitive"})

---

Base Classes

BaseSecurityDecorator

guard_core.decorators.base.BaseSecurityDecorator(config)

Source code in guard_core/decorators/base.py

def __init__(self, config: SecurityConfig) -> None:
    from guard_core.handlers.behavior_handler import BehaviorTracker

    self.config = config
    self._route_configs: dict[str, RouteConfig] = {}
    # id -> id(func) of the function the id was derived from. A stamped
    # _guard_route_id is only reused when THIS instance assigned it to
    # THIS function; otherwise a cross-instance collision would select
    # another route's config.
    self._route_id_owner: dict[str, int] = {}
    self._route_config_revision = RouteConfigRevision()
    self.behavior_tracker = BehaviorTracker(config)
    self.agent_handler: Any = None
    self.geo_ip_handler: Any = None

agent_handler = None instance-attribute

behavior_tracker = BehaviorTracker(config) instance-attribute

config = config instance-attribute

geo_ip_handler = None instance-attribute

route_config_revision property

get_route_config(route_id)

Source code in guard_core/decorators/base.py

def get_route_config(self, route_id: str) -> RouteConfig | None:
    return self._route_configs.get(route_id)

initialize_agent(agent_handler, geo_ip_handler=None) async

Source code in guard_core/decorators/base.py

async def initialize_agent(
    self, agent_handler: Any, geo_ip_handler: Any = None
) -> None:
    self.agent_handler = agent_handler
    self.geo_ip_handler = geo_ip_handler
    await self.behavior_tracker.initialize_agent(agent_handler)

initialize_behavior_tracking(redis_handler=None) async

Source code in guard_core/decorators/base.py

async def initialize_behavior_tracking(self, redis_handler: Any = None) -> None:
    if redis_handler:
        await self.behavior_tracker.initialize_redis(redis_handler)

send_access_denied_event(request, reason, decorator_type, **metadata) async

Source code in guard_core/decorators/base.py

async def send_access_denied_event(
    self,
    request: GuardRequest,
    reason: str,
    decorator_type: str,
    **metadata: Any,
) -> None:
    from guard_core.core.events.event_types import EVENT_ACCESS_DENIED

    await self.send_decorator_event(
        event_type=EVENT_ACCESS_DENIED,
        request=request,
        action_taken="blocked",
        reason=reason,
        decorator_type=decorator_type,
        **metadata,
    )

send_authentication_failed_event(request, reason, auth_type, **metadata) async

Source code in guard_core/decorators/base.py

async def send_authentication_failed_event(
    self,
    request: GuardRequest,
    reason: str,
    auth_type: str,
    **metadata: Any,
) -> None:
    from guard_core.core.events.event_types import EVENT_AUTHENTICATION_FAILED

    await self.send_decorator_event(
        event_type=EVENT_AUTHENTICATION_FAILED,
        request=request,
        action_taken="blocked",
        reason=reason,
        decorator_type="authentication",
        auth_type=auth_type,
        **metadata,
    )

send_decorator_event(event_type, request, action_taken, reason, decorator_type, **kwargs) async

Source code in guard_core/decorators/base.py

async def send_decorator_event(
    self,
    event_type: str,
    request: GuardRequest,
    action_taken: str,
    reason: str,
    decorator_type: str,
    **kwargs: Any,
) -> None:
    if not self.agent_handler:
        return

    from guard_core.core.events.middleware_events import SecurityEventBus

    event_bus = SecurityEventBus(
        agent_handler=self.agent_handler,
        config=self.config,
        geo_ip_handler=self.geo_ip_handler,
    )
    await event_bus.send_middleware_event(
        event_type=event_type,
        request=request,
        action_taken=action_taken,
        reason=reason,
        decorator_type=decorator_type,
        **kwargs,
    )

send_decorator_violation_event(request, violation_type, reason, **metadata) async

Source code in guard_core/decorators/base.py

async def send_decorator_violation_event(
    self,
    request: GuardRequest,
    violation_type: str,
    reason: str,
    **metadata: Any,
) -> None:
    from guard_core.core.events.event_types import EVENT_DECORATOR_VIOLATION

    await self.send_decorator_event(
        event_type=EVENT_DECORATOR_VIOLATION,
        request=request,
        action_taken="blocked",
        reason=reason,
        decorator_type=violation_type,
        **metadata,
    )

send_rate_limit_event(request, limit, window, **metadata) async

Source code in guard_core/decorators/base.py

async def send_rate_limit_event(
    self,
    request: GuardRequest,
    limit: int,
    window: int,
    **metadata: Any,
) -> None:
    from guard_core.core.events.event_types import EVENT_RATE_LIMITED

    await self.send_decorator_event(
        event_type=EVENT_RATE_LIMITED,
        request=request,
        action_taken="blocked",
        reason=f"Rate limit exceeded: {limit} requests per {window}s",
        decorator_type="rate_limiting",
        limit=limit,
        window=window,
        **metadata,
    )

Base class providing core decorator functionality and route configuration management.

RouteConfig

guard_core.decorators.base.RouteConfig(revision=None)

Source code in guard_core/decorators/route_config.py

def __init__(self, revision: RouteConfigRevision | None = None) -> None:
    object.__setattr__(self, "_revision", revision)
    object.__setattr__(self, "_initialized", False)
    self.rate_limit: int | None = None
    self.rate_limit_window: int | None = None
    self.ip_whitelist: list[str] | None = None
    self.ip_blacklist: list[str] | None = None
    self.blocked_countries: list[str] | None = None
    self.whitelist_countries: list[str] | None = None
    self.bypassed_checks: set[str] = set()
    self.require_https: bool = False
    self.auth_required: str | None = None
    self.custom_validators: list[Callable] = []
    self.blocked_user_agents: list[str] = []
    self.required_headers: dict[str, str] = {}
    self.behavior_rules: list[BehaviorRule] = []
    self.block_cloud_providers: set[str] = set()
    self.max_request_size: int | None = None
    self.allowed_content_types: list[str] | None = None
    self.time_restrictions: dict[str, str] | None = None
    self.enable_suspicious_detection: bool = True
    self.require_referrer: list[str] | None = None
    self.api_key_required: bool = False
    self.auth_verifier: AuthVerifier | None = None
    self.api_key_verifier: AuthVerifier | None = None
    self.api_key_header: str | None = None
    self.authorization_header_required: str | None = None
    self.geo_rate_limits: dict[str, tuple[int, int]] | None = None
    self.excluded_detection_headers: set[str] | None = None
    self.excluded_detection_params: set[str] | None = None
    self.excluded_detection_body_fields: set[str] | None = None
    self.enabled_detection_categories: set[str] | None = None
    self.detection_scan_body: bool | None = None
    object.__setattr__(self, "_initialized", True)

allowed_content_types = None instance-attribute

api_key_header = None instance-attribute

api_key_required = False instance-attribute

api_key_verifier = None instance-attribute

auth_required = None instance-attribute

auth_verifier = None instance-attribute

authorization_header_required = None instance-attribute

behavior_rules = [] instance-attribute

block_cloud_providers = set() instance-attribute

blocked_countries = None instance-attribute

blocked_user_agents = [] instance-attribute

bypassed_checks = set() instance-attribute

custom_validators = [] instance-attribute

detection_scan_body = None instance-attribute

enable_suspicious_detection = True instance-attribute

enabled_detection_categories = None instance-attribute

excluded_detection_body_fields = None instance-attribute

excluded_detection_headers = None instance-attribute

excluded_detection_params = None instance-attribute

geo_rate_limits = None instance-attribute

ip_blacklist = None instance-attribute

ip_whitelist = None instance-attribute

max_request_size = None instance-attribute

rate_limit = None instance-attribute

rate_limit_window = None instance-attribute

require_https = False instance-attribute

require_referrer = None instance-attribute

required_headers = {} instance-attribute

time_restrictions = None instance-attribute

whitelist_countries = None instance-attribute

Configuration class that stores security settings for individual routes.

---

Mixin Classes

The decorator system uses mixins to organize different types of security features:

AccessControlMixin

guard_core.decorators.access_control.AccessControlMixin

Bases: BaseSecurityMixin

allow_countries(countries)

Source code in guard_core/decorators/access_control.py

def allow_countries(
    self, countries: list[str]
) -> Callable[[Callable[..., Any]], DecoratedFunction]:
    def decorator(func: Callable[..., Any]) -> DecoratedFunction:
        route_config = self._ensure_route_config(func)
        route_config.whitelist_countries = [c.upper() for c in countries]
        return self._apply_route_config(func)

    return decorator

block_clouds(providers=None)

Source code in guard_core/decorators/access_control.py

def block_clouds(
    self, providers: list[str] | None = None
) -> Callable[[Callable[..., Any]], DecoratedFunction]:
    def decorator(func: Callable[..., Any]) -> DecoratedFunction:
        route_config = self._ensure_route_config(func)
        if providers is None:
            route_config.block_cloud_providers = set(VALID_CLOUD_PROVIDERS)
        else:
            valid = {
                p
                for p in providers
                if p.partition(":!")[0] in VALID_CLOUD_PROVIDERS
            }
            route_config.block_cloud_providers = valid
            invalid = set(providers) - valid
            if invalid:
                logging.getLogger("guard_core.decorators").warning(
                    "@block_clouds: ignored unknown cloud providers %s",
                    sorted(invalid),
                )
        return self._apply_route_config(func)

    return decorator

block_countries(countries)

Source code in guard_core/decorators/access_control.py

def block_countries(
    self, countries: list[str]
) -> Callable[[Callable[..., Any]], DecoratedFunction]:
    def decorator(func: Callable[..., Any]) -> DecoratedFunction:
        route_config = self._ensure_route_config(func)
        route_config.blocked_countries = [c.upper() for c in countries]
        return self._apply_route_config(func)

    return decorator

bypass(checks)

Source code in guard_core/decorators/access_control.py

def bypass(
    self, checks: list[str]
) -> Callable[[Callable[..., Any]], DecoratedFunction]:
    def decorator(func: Callable[..., Any]) -> DecoratedFunction:
        route_config = self._ensure_route_config(func)
        valid = {c for c in checks if c in VALID_BYPASS_CHECKS}
        route_config.bypassed_checks.update(valid)
        invalid = set(checks) - valid
        if invalid:
            logging.getLogger("guard_core.decorators").warning(
                "@bypass: ignored unknown checks %s",
                sorted(invalid, key=str),
            )
        return self._apply_route_config(func)

    return decorator

require_ip(whitelist=None, blacklist=None)

Source code in guard_core/decorators/access_control.py

def require_ip(
    self,
    whitelist: list[str] | None = None,
    blacklist: list[str] | None = None,
) -> Callable[[Callable[..., Any]], DecoratedFunction]:
    def decorator(func: Callable[..., Any]) -> DecoratedFunction:
        route_config = self._ensure_route_config(func)
        if whitelist:
            route_config.ip_whitelist = whitelist
        if blacklist:
            route_config.ip_blacklist = blacklist
        return self._apply_route_config(func)

    return decorator

Provides IP-based and geographic access control decorators.

Available Decorators:

• @guard_deco.require_ip(whitelist=[], blacklist=[]) - IP address filtering
• @guard_deco.block_countries(countries=[]) - Block specific countries
• @guard_deco.allow_countries(countries=[]) - Allow only specific countries
• @guard_deco.block_clouds(providers=[]) - Block cloud provider IPs
• @guard_deco.bypass(checks=[]) - Bypass specific security checks

Example:

class AccessIPWhitelistHandler(SecurityHandler):
    @guard_deco.require_ip(whitelist=["127.0.0.1", "10.0.0.0/8"])
    async def get(self) -> None:
        self.write({"message": "Access granted from whitelisted IP"})

AuthenticationMixin

guard_core.decorators.authentication.AuthenticationMixin

Bases: BaseSecurityMixin

api_key_auth(header_name='X-API-Key', verifier=None)

Source code in guard_core/decorators/authentication.py

def api_key_auth(
    self,
    header_name: str = "X-API-Key",
    verifier: AuthVerifier | None = None,
) -> Callable[[Callable[..., Any]], DecoratedFunction]:
    def decorator(func: Callable[..., Any]) -> DecoratedFunction:
        route_config = self._ensure_route_config(func)
        route_config.api_key_required = True
        route_config.required_headers[header_name] = "required"
        route_config.api_key_header = header_name
        route_config.api_key_verifier = verifier
        if route_config.authorization_header_required is not None:
            raise ValueError(
                "api_key_auth cannot be combined with require_authorization_header;"
                " the latter is presence-only and mutually exclusive with"
                " authenticated routes"
            )
        return self._apply_route_config(func)

    return decorator

require_auth(type='bearer', verifier=None)

Source code in guard_core/decorators/authentication.py

def require_auth(
    self,
    type: str = "bearer",
    verifier: AuthVerifier | None = None,
) -> Callable[[Callable[..., Any]], DecoratedFunction]:
    def decorator(func: Callable[..., Any]) -> DecoratedFunction:
        route_config = self._ensure_route_config(func)
        route_config.auth_required = type
        route_config.auth_verifier = verifier
        if route_config.authorization_header_required is not None:
            raise ValueError(
                "require_auth cannot be combined with require_authorization_header;"
                " the latter is presence-only and mutually exclusive with"
                " authenticated routes"
            )
        return self._apply_route_config(func)

    return decorator

require_authorization_header(scheme='bearer')

Source code in guard_core/decorators/authentication.py

def require_authorization_header(
    self,
    scheme: str = "bearer",
) -> Callable[[Callable[..., Any]], DecoratedFunction]:
    def decorator(func: Callable[..., Any]) -> DecoratedFunction:
        route_config = self._ensure_route_config(func)
        route_config.authorization_header_required = scheme
        if route_config.auth_required is not None or route_config.api_key_required:
            raise ValueError(
                "require_authorization_header cannot be combined with"
                " require_auth or api_key_auth; it is presence-only and"
                " mutually exclusive with authenticated routes"
            )
        return self._apply_route_config(func)

    return decorator

require_headers(headers)

Source code in guard_core/decorators/authentication.py

def require_headers(
    self, headers: dict[str, str]
) -> Callable[[Callable[..., Any]], DecoratedFunction]:
    def decorator(func: Callable[..., Any]) -> DecoratedFunction:
        route_config = self._ensure_route_config(func)
        route_config.required_headers.update(headers)
        return self._apply_route_config(func)

    return decorator

require_https()

Source code in guard_core/decorators/authentication.py

def require_https(self) -> Callable[[Callable[..., Any]], DecoratedFunction]:
    def decorator(func: Callable[..., Any]) -> DecoratedFunction:
        route_config = self._ensure_route_config(func)
        route_config.require_https = True
        return self._apply_route_config(func)

    return decorator

Provides authentication and authorization decorators.

Available Decorators:

• @guard_deco.require_https() - Force HTTPS
• @guard_deco.require_auth(type="bearer") - Require authentication
• @guard_deco.api_key_auth(header_name="X-API-Key") - API key authentication
• @guard_deco.require_headers(headers={}) - Require specific headers

Example:

class AuthBearerHandler(SecurityHandler):
    @guard_deco.require_auth(type="bearer")
    async def get(self) -> None:
        self.write({"authenticated": True})

RateLimitingMixin

guard_core.decorators.rate_limiting.RateLimitingMixin

Bases: BaseSecurityMixin

geo_rate_limit(limits)

Source code in guard_core/decorators/rate_limiting.py

def geo_rate_limit(
    self, limits: dict[str, tuple[int, int]]
) -> Callable[[Callable[..., Any]], DecoratedFunction]:
    def decorator(func: Callable[..., Any]) -> DecoratedFunction:
        route_config = self._ensure_route_config(func)
        route_config.geo_rate_limits = limits
        return self._apply_route_config(func)

    return decorator

rate_limit(requests, window=60)

Source code in guard_core/decorators/rate_limiting.py

def rate_limit(
    self, requests: int, window: int = 60
) -> Callable[[Callable[..., Any]], DecoratedFunction]:
    def decorator(func: Callable[..., Any]) -> DecoratedFunction:
        route_config = self._ensure_route_config(func)
        route_config.rate_limit = requests
        route_config.rate_limit_window = window
        return self._apply_route_config(func)

    return decorator

Provides rate limiting decorators.

Available Decorators:

• @guard_deco.rate_limit(requests=10, window=60) - Basic rate limiting
• @guard_deco.geo_rate_limit(limits={}) - Geographic rate limiting

Example:

class RateCustomLimitHandler(SecurityHandler):
    @guard_deco.rate_limit(requests=5, window=60)
    async def get(self) -> None:
        self.write({"message": "rate limited"})

BehavioralMixin

guard_core.decorators.behavioral.BehavioralMixin

Bases: BaseSecurityMixin

config instance-attribute

behavior_analysis(rules)

Source code in guard_core/decorators/behavioral.py

def behavior_analysis(
    self, rules: list[BehaviorRule]
) -> Callable[[Callable[..., Any]], DecoratedFunction]:
    for rule in rules:
        if rule.rule_type == "return_pattern" and rule.pattern:
            _validate_return_pattern_body_scan(rule.pattern, self.config)

    def decorator(func: Callable[..., Any]) -> DecoratedFunction:
        route_config = self._ensure_route_config(func)
        route_config.behavior_rules.extend(rules)
        return self._apply_route_config(func)

    return decorator

return_monitor(pattern, max_occurrences, window=86400, action='ban')

Source code in guard_core/decorators/behavioral.py

def return_monitor(
    self,
    pattern: str,
    max_occurrences: int,
    window: int = 86400,
    action: Literal["ban", "log", "throttle", "alert"] = "ban",
) -> Callable[[Callable[..., Any]], DecoratedFunction]:
    _validate_return_pattern_body_scan(pattern, self.config)

    def decorator(func: Callable[..., Any]) -> DecoratedFunction:
        route_config = self._ensure_route_config(func)

        rule = BehaviorRule(
            rule_type="return_pattern",
            threshold=max_occurrences,
            window=window,
            pattern=pattern,
            action=action,
        )
        route_config.behavior_rules.append(rule)
        return self._apply_route_config(func)

    return decorator

suspicious_frequency(max_frequency, window=300, action='ban')

Source code in guard_core/decorators/behavioral.py

def suspicious_frequency(
    self,
    max_frequency: float,
    window: int = 300,
    action: Literal["ban", "log", "throttle", "alert"] = "ban",
) -> Callable[[Callable[..., Any]], DecoratedFunction]:
    def decorator(func: Callable[..., Any]) -> DecoratedFunction:
        route_config = self._ensure_route_config(func)
        max_calls = int(max_frequency * window)

        rule = BehaviorRule(
            rule_type="frequency",
            threshold=max_calls,
            window=window,
            action=action,
        )
        route_config.behavior_rules.append(rule)
        return self._apply_route_config(func)

    return decorator

usage_monitor(max_calls, window=3600, action='ban')

Source code in guard_core/decorators/behavioral.py

def usage_monitor(
    self,
    max_calls: int,
    window: int = 3600,
    action: Literal["ban", "log", "throttle", "alert"] = "ban",
) -> Callable[[Callable[..., Any]], DecoratedFunction]:
    def decorator(func: Callable[..., Any]) -> DecoratedFunction:
        route_config = self._ensure_route_config(func)

        rule = BehaviorRule(
            rule_type="usage", threshold=max_calls, window=window, action=action
        )
        route_config.behavior_rules.append(rule)
        return self._apply_route_config(func)

    return decorator

Provides behavioral analysis and monitoring decorators.

Available Decorators:

• @guard_deco.usage_monitor(max_calls, window, action) - Monitor endpoint usage
• @guard_deco.return_monitor(pattern, max_occurrences, window, action) - Monitor return patterns
• @guard_deco.behavior_analysis(rules=[]) - Apply multiple behavioral rules
• @guard_deco.suspicious_frequency(max_frequency, window, action) - Detect suspicious frequency

Example:

class BehaviorUsageHandler(SecurityHandler):
    @guard_deco.usage_monitor(max_calls=10, window=300, action="log")
    async def get(self) -> None:
        self.write({"message": "Usage monitoring active"})

ContentFilteringMixin

guard_core.decorators.content_filtering.ContentFilteringMixin

Bases: BaseSecurityMixin

config instance-attribute

block_user_agents(patterns)

Source code in guard_core/decorators/content_filtering.py

def block_user_agents(
    self, patterns: list[str]
) -> Callable[[Callable[..., Any]], DecoratedFunction]:
    from guard_core.detection_engine.compiler import PatternCompiler

    compiler = PatternCompiler()
    for pattern in patterns:
        is_safe, reason = compiler.validate_pattern_safety(
            pattern,
            max_content_length=self.config.detection_max_body_inspect_bytes,
        )
        if not is_safe:
            from guard_core._utils.detection_scan import _redact_pattern_source

            raise ValueError(
                f"block_user_agents pattern rejected by ReDoS validator: "
                f"{_redact_pattern_source(pattern)!r} ({reason})"
            )

    def decorator(func: Callable[..., Any]) -> DecoratedFunction:
        route_config = self._ensure_route_config(func)
        route_config.blocked_user_agents.extend(patterns)
        return self._apply_route_config(func)

    return decorator

content_type_filter(allowed_types)

Source code in guard_core/decorators/content_filtering.py

def content_type_filter(
    self, allowed_types: list[str]
) -> Callable[[Callable[..., Any]], DecoratedFunction]:
    def decorator(func: Callable[..., Any]) -> DecoratedFunction:
        route_config = self._ensure_route_config(func)
        route_config.allowed_content_types = allowed_types
        return self._apply_route_config(func)

    return decorator

custom_validation(validator)

Source code in guard_core/decorators/content_filtering.py

def custom_validation(
    self,
    validator: Callable[[GuardRequest], Awaitable[GuardResponse | None]],
) -> Callable[[Callable[..., Any]], DecoratedFunction]:
    def decorator(func: Callable[..., Any]) -> DecoratedFunction:
        route_config = self._ensure_route_config(func)
        route_config.custom_validators.append(validator)
        return self._apply_route_config(func)

    return decorator

detection_exclusion(headers=None, params=None, body_fields=None, categories=None, scan_body=None)

Source code in guard_core/decorators/content_filtering.py

def detection_exclusion(
    self,
    headers: set[str] | None = None,
    params: set[str] | None = None,
    body_fields: set[str] | None = None,
    categories: set[str] | None = None,
    scan_body: bool | None = None,
) -> Callable[[Callable[..., Any]], DecoratedFunction]:
    def decorator(func: Callable[..., Any]) -> DecoratedFunction:
        route_config = self._ensure_route_config(func)
        if headers is not None:
            route_config.excluded_detection_headers = set(headers)
        if params is not None:
            route_config.excluded_detection_params = set(params)
        if body_fields is not None:
            route_config.excluded_detection_body_fields = set(body_fields)
        if categories is not None:
            route_config.enabled_detection_categories = set(categories)
        if scan_body is not None:
            route_config.detection_scan_body = scan_body
        return self._apply_route_config(func)

    return decorator

max_request_size(size_bytes)

Source code in guard_core/decorators/content_filtering.py

def max_request_size(
    self, size_bytes: int
) -> Callable[[Callable[..., Any]], DecoratedFunction]:
    def decorator(func: Callable[..., Any]) -> DecoratedFunction:
        route_config = self._ensure_route_config(func)
        route_config.max_request_size = size_bytes
        return self._apply_route_config(func)

    return decorator

require_referrer(allowed_domains)

Source code in guard_core/decorators/content_filtering.py

def require_referrer(
    self, allowed_domains: list[str]
) -> Callable[[Callable[..., Any]], DecoratedFunction]:
    def decorator(func: Callable[..., Any]) -> DecoratedFunction:
        route_config = self._ensure_route_config(func)
        route_config.require_referrer = allowed_domains
        return self._apply_route_config(func)

    return decorator

Provides content and request filtering decorators.

Available Decorators:

• @guard_deco.block_user_agents(patterns=[]) - Block user agent patterns
• @guard_deco.content_type_filter(allowed_types=[]) - Filter content types
• @guard_deco.max_request_size(size_bytes) - Limit request size
• @guard_deco.require_referrer(allowed_domains=[]) - Require specific referrers
• @guard_deco.custom_validation(validator) - Add custom validation logic

Example:

class ContentJSONOnlyHandler(SecurityHandler):
    @guard_deco.content_type_filter(["application/json"])
    async def post(self) -> None:
        self.write({"message": "JSON content received"})

AdvancedMixin

guard_core.decorators.advanced.AdvancedMixin

Bases: BaseSecurityMixin

config instance-attribute

honeypot_detection(trap_fields)

Source code in guard_core/decorators/advanced.py

def honeypot_detection(
    self, trap_fields: list[str]
) -> Callable[[Callable[..., Any]], DecoratedFunction]:
    config = self.config

    def decorator(func: Callable[..., Any]) -> DecoratedFunction:
        async def honeypot_validator(
            request: GuardRequest,
        ) -> GuardResponse | None:
            def _has_trap_field_filled(data: dict[str, Any]) -> bool:
                return any(field in data and data[field] for field in trap_fields)

            def _validate_form_data(raw_body: bytes) -> GuardResponse | None:
                parsed = parse_qs(
                    raw_body.decode("utf-8", errors="surrogateescape")
                )
                flat = {k: v[0] for k, v in parsed.items() if v}
                if _has_trap_field_filled(flat):
                    return _SimpleResponse("Forbidden", 403)
                return None

            def _validate_json_data(raw_body: bytes) -> GuardResponse | None:
                try:
                    json_data = json.loads(
                        raw_body.decode("utf-8", errors="surrogateescape")
                    )
                    if _has_trap_field_filled(json_data):
                        return _SimpleResponse("Forbidden", 403)
                except (json.JSONDecodeError, TypeError, RecursionError) as exc:
                    logger.debug(
                        "honeypot_detection: skipping unparsable JSON body: %s",
                        exc,
                    )
                return None

            if request.method not in ["POST", "PUT", "PATCH"]:
                return None

            raw_body = await _read_capped_body(request, config)
            if raw_body is None:
                return None

            content_type = request.headers.get("content-type", "")

            if "application/x-www-form-urlencoded" in content_type:
                return _validate_form_data(raw_body)
            elif "application/json" in content_type:
                return _validate_json_data(raw_body)

            return None

        route_config = self._ensure_route_config(func)
        route_config.custom_validators.append(honeypot_validator)
        return self._apply_route_config(func)

    return decorator

suspicious_detection(enabled=True)

Source code in guard_core/decorators/advanced.py

def suspicious_detection(
    self, enabled: bool = True
) -> Callable[[Callable[..., Any]], DecoratedFunction]:
    def decorator(func: Callable[..., Any]) -> DecoratedFunction:
        route_config = self._ensure_route_config(func)
        route_config.enable_suspicious_detection = enabled
        return self._apply_route_config(func)

    return decorator

time_window(start_time, end_time, timezone='UTC')

Source code in guard_core/decorators/advanced.py

def time_window(
    self, start_time: str, end_time: str, timezone: str = "UTC"
) -> Callable[[Callable[..., Any]], DecoratedFunction]:
    def decorator(func: Callable[..., Any]) -> DecoratedFunction:
        route_config = self._ensure_route_config(func)
        route_config.time_restrictions = {
            "start": start_time,
            "end": end_time,
            "timezone": timezone,
        }
        return self._apply_route_config(func)

    return decorator

Provides advanced detection and time-based decorators.

Available Decorators:

• @guard_deco.time_window(start_time, end_time, timezone) - Time-based access control
• @guard_deco.suspicious_detection(enabled=True) - Toggle suspicious pattern detection
• @guard_deco.honeypot_detection(trap_fields=[]) - Detect bots using honeypot fields

Example:

class AdvancedBusinessHoursHandler(SecurityHandler):
    @guard_deco.time_window(start_time="09:00", end_time="17:00", timezone="UTC")
    async def get(self) -> None:
        self.write({"message": "Access granted during business hours"})

---

Utility Functions

. get_route_decorator_config

guard_core.decorators.base.get_route_decorator_config(request, decorator_handler)

Source code in guard_core/decorators/base.py

def get_route_decorator_config(
    request: GuardRequest, decorator_handler: BaseSecurityDecorator
) -> RouteConfig | None:
    route_id = getattr(request.state, "guard_route_id", None)
    if route_id:
        return decorator_handler.get_route_config(route_id)
    return None

Extract route security configuration from the current request. SecurityMiddleware invokes this internally via RouteConfigResolver.get_route_config() to read the decorator metadata attached to the currently dispatched handler method.

---

Integration with Middleware

The decorators work in conjunction with SecurityMiddleware to provide comprehensive protection:

1. Route Configuration — decorators attach metadata onto the handler method
2. Middleware Processing — SecurityMiddleware.run_pre_processing() reads that metadata and applies the corresponding checks
3. Override Behavior — route-specific settings override global middleware settings for the matched handler

Example Integration:

import asyncio

import tornado.web
from tornadoapi_guard import (
    SecurityConfig,
    SecurityDecorator,
    SecurityHandler,
    SecurityMiddleware,
)

config = SecurityConfig(
    enable_ip_banning=True,
    enable_rate_limiting=True,
    rate_limit_requests=100,
    rate_limit_window=3600,
)

middleware = SecurityMiddleware(config=config)
guard_deco = SecurityDecorator(config)
middleware.set_decorator_handler(guard_deco)


class LimitedHandler(SecurityHandler):
    @guard_deco.rate_limit(requests=10, window=300)  # Override: 10 requests/5min
    async def get(self) -> None:
        self.write({"data": "limited"})


class PublicHandler(SecurityHandler):
    async def get(self) -> None:
        self.write({"data": "public"})


async def main() -> None:
    await middleware.initialize()
    app = tornado.web.Application(
        [
            (r"/api/limited", LimitedHandler),
            (r"/api/public", PublicHandler),
        ],
        security_middleware=middleware,
        guard_decorator=guard_deco,
    )
    app.listen(8000)
    await asyncio.Event().wait()


if __name__ == "__main__":
    asyncio.run(main())

---

Best Practices

Decorator Order

Apply decorators in logical order, with more specific restrictions first:

class AdminSensitiveHandler(SecurityHandler):
    @guard_deco.require_https()                        # Security requirement
    @guard_deco.require_auth(type="bearer")            # Authentication
    @guard_deco.require_ip(whitelist=["10.0.0.0/8"])   # Access control
    @guard_deco.rate_limit(requests=5, window=3600)    # Rate limiting
    @guard_deco.suspicious_detection(enabled=True)     # Monitoring
    async def post(self) -> None:
        self.write({"status": "admin action"})

Combining Behavioral Analysis

Use multiple behavioral decorators for comprehensive monitoring:

class RewardsHandler(SecurityHandler):
    @guard_deco.usage_monitor(max_calls=50, window=3600, action="ban")
    @guard_deco.return_monitor("rare_item", max_occurrences=3, window=86400, action="ban")
    @guard_deco.suspicious_frequency(max_frequency=0.1, window=300, action="alert")
    async def get(self) -> None:
        self.write({"reward": "rare_item", "value": 1000})

Geographic and Cloud Controls

Combine geographic and cloud provider controls:

class RestrictedHandler(SecurityHandler):
    @guard_deco.allow_countries(["US", "CA", "GB"])
    @guard_deco.block_clouds(["AWS", "GCP"])
    async def get(self) -> None:
        self.write({"data": "geo-restricted"})

Content Filtering

Apply content filtering for upload endpoints:

class UploadHandler(SecurityHandler):
    @guard_deco.content_type_filter(["image/jpeg", "image/png"])
    @guard_deco.max_request_size(5 * 1024 * 1024)
    @guard_deco.require_referrer(["myapp.com"])
    async def post(self) -> None:
        self.write({"status": "uploaded"})

---

Error Handling

Decorators integrate with the middleware's error handling system. When decorator conditions are not met, appropriate HTTP responses are returned:

. 403 Forbidden

IP restrictions, country blocks, authentication failures

. 429 Too Many Requests

Rate limiting violations

. 400 Bad Request

Content type mismatches, missing headers

. 413 Payload Too Large

Request size limits exceeded

---

Configuration Priority

Security settings are applied in the following priority order:

1. Decorator Settings (highest priority)
2. Global Middleware Settings
3. Default Settings (lowest priority)

This allows for flexible override behavior where routes can customize their security requirements while maintaining global defaults.