langchain-core: KeyError DoS via Malformed Content Blocks

from ark_trust import OutputValidator, Schema

# 定义 MessageSchema，确保 content_blocks 结构完整
message_schema = Schema(
    type="object",
    fields=[
        Schema.Field("type", "string", required=True),
        Schema.Field("source", "object", required=True),
    ],
    allow_extra=False
)

validator = OutputValidator(schema=message_schema)

# 外部消息入队前校验
raw_message = get_external_message()  # 来自 Webhook / 用户 / 工具
result = validator.validate(raw_message.content_blocks)
if not result.valid:
    raise ValidationError(f"Malformed content_blocks rejected: {result.errors}")
    # 进程存活，错误可追踪，不影响其他请求

from ark_trust import CircuitBreaker

# 外部输入源连续失败时熔断
breaker = CircuitBreaker(
    failure_threshold=3,    # 3 次 KeyError → 打开
    recovery_timeout=60,    # 60 秒后尝试半开
)

async def process_external_message(msg):
    try:
        blocks = msg.content_blocks  # 可能抛 KeyError
    except KeyError:
        breaker.record_failure()
        raise ExternalInputError(msg)  # 快速失败，不拖垮 Agent
    else:
        breaker.record_success()
        return blocks

# breaker.open() 后，所有外部消息直接拒绝
# 保护 Agent 进程不被外部畸形数据 DoS

from ark_trust import IdempotencyGuard

guard = IdempotencyGuard(ttl=300)

# 畸形消息被第一次拦截后，相同 key 直接命中缓存
result = guard.execute(
    key=msg.id,
    fn=lambda: process_external_message(msg),
)
# 第二次相同畸形消息：缓存命中 → 零额外处理
# 防止攻击者重放同一畸形消息消耗资源