AID-H-002.008: Realtime Multimodal Input Stream Security

Version 1.20261015 · Actionable sub-technique

Safely decode, normalize, segment, and admit realtime multimodal streams before transcripts, OCR text, visual cues, audio embeddings, or live media events are fused into an AI application or agent workflow. This sub-technique covers live microphone audio, voice-mode sessions, camera/video streams, video-call feeds, and screen-share streams where the system must make incremental decisions before the full artifact exists. Deterministic controls include bounded voice-activity turns, frame sampling, transcript/OCR normalization, control-token rejection, and fresh admission records.

Pillars: data, app

Lifecycle: building, operation

Scope boundary

Owns safe decoding, normalization, segmentation, bounded sampling, control-token handling, and the preventive admit, block, or downgrade decision for realtime microphone, camera, call, video, and screen-share streams.

AID-H-002.003: AID-H-002.008 gates ephemeral streams incrementally; AID-H-002.003 sanitizes complete non-realtime files and artifacts. Live timing, transcript churn, and source changes require decisions before stable artifact bytes exist.

AID-D-007.002: AID-H-002.008 owns preventive realtime stream admission and downgrade; AID-D-007.002 emits semantic, temporal, source-change, and action-timing findings over the normalized stream. The detector informs policy, while this Harden control decides whether stream observations may reach the model or agent.

Implementation guidance

AID-H-002.008-G001

Normalize live audio to one canonical PCM profile, segment speech turns with bounded voice activity detection, and reject segments that exceed duration, confidence, or transcript-control-token policy.
Concept:

Audio is one realtime multimodal stream, not a special taxonomy category. Treat microphone and voice-mode input as an incremental stream that must be normalized, segmented, and gated before the transcript can become model instructions, tool parameters, or memory. The same admission contract should be used alongside camera, video, or screen-share gates in this sub-technique.

Step 1: Normalize and segment audio before transcription
# File: realtime_input/audio_gate.py
from __future__ import annotations
import os

import io
import re
from dataclasses import dataclass

import webrtcvad
from pydub import AudioSegment

def _required_positive_int(name: str) -> int:
    value = int(os.environ[name])
    if value <= 0:
        raise RuntimeError(f"{name} must be a positive integer")
    return value

SAMPLE_RATE = _required_positive_int("VERIFIED_H002_REALTIME_AUDIO_SAMPLE_RATE_HZ")
FRAME_MS = _required_positive_int("VERIFIED_H002_REALTIME_AUDIO_FRAME_MILLISECONDS")
MAX_SEGMENT_MS = _required_positive_int("VERIFIED_H002_REALTIME_AUDIO_MAX_SEGMENT_MILLISECONDS")
MAX_STREAM_MS = _required_positive_int("VERIFIED_H002_REALTIME_AUDIO_MAX_STREAM_MILLISECONDS")
VAD_MODE = _required_positive_int("VERIFIED_H002_REALTIME_AUDIO_VAD_MODE")



# WebRTC VAD supports only these named PCM protocol profiles.
if SAMPLE_RATE not in {8000, 16000, 32000, 48000} or FRAME_MS not in {10, 20, 30} or VAD_MODE not in {0, 1, 2, 3}:
    raise RuntimeError("verified realtime audio profile is unsupported by WebRTC VAD")

CONTROL_TOKEN_RE = re.compile(r"(<\|/?(?:system|developer|assistant|tool).*?\|>|\b(system|developer)\s*:)", re.I)


@dataclass(frozen=True)
class VoiceSegment:
    pcm16le: bytes
    start_ms: int
    end_ms: int


def normalize_audio(blob: bytes, input_format: str) -> AudioSegment:
    if not blob:
        raise ValueError("empty_audio_stream")
    source = AudioSegment.from_file(io.BytesIO(blob), format=input_format)
    if len(source) > MAX_STREAM_MS:
        raise ValueError("audio_stream_too_long")
    return source.set_frame_rate(SAMPLE_RATE).set_channels(1).set_sample_width(2)


def split_speech_segments(blob: bytes, input_format: str) -> list[VoiceSegment]:
    audio = normalize_audio(blob, input_format)
    vad = webrtcvad.Vad(VAD_MODE)
    frame_bytes = int(SAMPLE_RATE * FRAME_MS / 1000) * 2
    pcm = audio.raw_data

    segments: list[VoiceSegment] = []
    current = bytearray()
    current_start_ms: int | None = None

    for offset in range(0, len(pcm) - frame_bytes + 1, frame_bytes):
        frame = pcm[offset:offset + frame_bytes]
        frame_start_ms = int(offset / 2 / SAMPLE_RATE * 1000)
        speech = vad.is_speech(frame, SAMPLE_RATE)

        if speech:
            if current_start_ms is None:
                current_start_ms = frame_start_ms
            current.extend(frame)
            if (frame_start_ms - current_start_ms) > MAX_SEGMENT_MS:
                raise ValueError("voice_segment_too_long")
        elif current_start_ms is not None:
            segments.append(VoiceSegment(bytes(current), current_start_ms, frame_start_ms))
            current.clear()
            current_start_ms = None

    if current_start_ms is not None:
        end_ms = int(len(pcm) / 2 / SAMPLE_RATE * 1000)
        segments.append(VoiceSegment(bytes(current), current_start_ms, end_ms))

    if not segments:
        raise ValueError("no_speech_detected")
    return segments
Step 2: Gate the ASR transcript before model fusion
# File: realtime_input/transcript_gate.py
from __future__ import annotations
import os

from realtime_input.audio_gate import CONTROL_TOKEN_RE
from multimodal.stream_findings import admission_reasons

def _required_positive_int(name: str) -> int:
    value = int(os.environ[name])
    if value <= 0:
        raise RuntimeError(f"{name} must be a positive integer")
    return value

MAX_TRANSCRIPT_CHARS = _required_positive_int("VERIFIED_H002_REALTIME_TRANSCRIPT_MAX_CHARS")





def admit_voice_transcript(
    transcript: str,
    workflow_risk: str,
    expected_session_id: str,
    expected_turn_index: int,
    d007_finding: dict,
    expected_policy_version: str,
    expected_policy_sha256: str,
) -> dict:
    cleaned = transcript.replace("\x00", "").strip()
    reasons: list[str] = []

    if not cleaned:
        reasons.append("empty_transcript")
    if len(cleaned) > MAX_TRANSCRIPT_CHARS:
        reasons.append("transcript_too_long")
    if CONTROL_TOKEN_RE.search(cleaned):
        reasons.append("control_token_or_role_marker")
    reasons.extend(admission_reasons(
        d007_finding, observation={"transcript": transcript},
        observation_id=f"audio:{expected_turn_index}", modality="audio",
        session_id=expected_session_id, policy_version=expected_policy_version,
        policy_sha256=expected_policy_sha256))

    decision = "allow" if not reasons else "block"
    return {
        "decision": decision,
        "reasons": reasons,
        "normalized_transcript": cleaned if decision == "allow" else "",
    }
Operational notes

Action: Normalize and bound each speech turn, pass the normalized observation to AID-D-007.002, then require a valid finding plus deterministic transcript checks before admitting speech as instructions.

Consume the signed aidefend.stream-admission-finding.v1 record produced by AID-D-007.002-G003. The trusted receiver verifies the detector identity, signature and freshness; it supplies the expected policy from current server configuration. Audio binds the original transcript object; visual binds the complete VisualStreamFrame object before normalization. Never accept the finding or policy from a client request.

AID-H-002.008-G002

Gate realtime video and screen-share frames with bounded frame sampling, OCR normalization, context allowlists, and high-risk action downgrade before visual stream content enters prompt or tool context.
Concept:

Realtime camera, live video, and screen-share streams are not ordinary uploaded images. The system sees them incrementally, and the dangerous text may appear for only a few frames before a tool action is proposed. Sample frames at a bounded rate, extract OCR text only when needed, and fail closed when visual context changes during high-impact actions.

Step 1: Normalize frame observations into an admission record
# File: realtime_input/visual_stream_gate.py
from __future__ import annotations
import os

import re
from dataclasses import dataclass, asdict
from multimodal.stream_findings import admission_reasons

def _required_positive_int(name: str) -> int:
    value = int(os.environ[name])
    if value <= 0:
        raise RuntimeError(f"{name} must be a positive integer")
    return value

MAX_FRAME_PIXELS = _required_positive_int("VERIFIED_H002_REALTIME_FRAME_MAX_PIXELS")
MAX_OCR_CHARS = _required_positive_int("VERIFIED_H002_REALTIME_OCR_MAX_CHARS")



CONTROL_TOKEN_RE = re.compile(r"(<\|/?(?:system|developer|assistant|tool).*?\|>|\b(system|developer)\s*:)", re.I)
ALLOWED_STREAM_TYPES = {"camera", "live_video", "screen_share"}


@dataclass(frozen=True)
class VisualStreamFrame:
    session_id: str
    frame_index: int
    stream_type: str
    width: int
    height: int
    ocr_text: str
    visual_confidence: float
    context_label: str
    workflow_risk: str


def admit_visual_frame(
    frame: VisualStreamFrame,
    allowed_context_labels: set[str],
    d007_finding: dict,
    expected_policy_version: str,
    expected_policy_sha256: str,
) -> dict:
    reasons: list[str] = []
    normalized_ocr = re.sub(r"\s+", " ", frame.ocr_text).strip()

    if frame.stream_type not in ALLOWED_STREAM_TYPES:
        reasons.append("stream_type_not_allowed")
    if frame.width * frame.height > MAX_FRAME_PIXELS:
        reasons.append("frame_too_large")
    if len(normalized_ocr) > MAX_OCR_CHARS:
        reasons.append("ocr_text_too_long")
    if CONTROL_TOKEN_RE.search(normalized_ocr):
        reasons.append("control_token_or_role_marker")
    if frame.context_label not in allowed_context_labels:
        reasons.append("visual_context_not_allowed")
    reasons.extend(admission_reasons(
        d007_finding, observation=asdict(frame),
        observation_id=f"visual:{frame.frame_index}", modality="visual",
        session_id=frame.session_id, policy_version=expected_policy_version,
        policy_sha256=expected_policy_sha256))

    decision = "allow" if not reasons else "block"
    return {
        "decision": decision,
        "reasons": reasons,
        "session_id": frame.session_id,
        "frame_index": frame.frame_index,
        "normalized_ocr": normalized_ocr if decision == "allow" else "",
        "context_label": frame.context_label,
    }
Step 2: Bind stream findings and gates to action timing

Send each normalized frame observation to AID-D-007.002. If a tool call, payment, message send, file export, or identity action is proposed after a visual context switch, require a fresh D-007 finding bound to the same session/frame and a fresh allowed admission decision before passing visual text into the action planner. Never reuse a result from an earlier screen, camera angle, tab, or participant.

Verification: Replay the complete sampled-frame population, signed admission policy, D-007 visual finding, OCR normalization output, and final decision; confirm the assembled context excludes every denied frame.

Action: Treat realtime visual content as per-frame preprocessing facts, D-007 findings, and H-002.008 admission records. Bound OCR and frame dimensions locally, consume the semantic detector finding, and force high-impact actions to use only fresh, exactly bound allow decisions.

Consume the signed aidefend.stream-admission-finding.v1 record produced by AID-D-007.002-G003. The trusted receiver verifies the detector identity, signature and freshness; it supplies the expected policy from current server configuration. Audio binds the original transcript object; visual binds the complete VisualStreamFrame object before normalization. Never accept the finding or policy from a client request.

Open-source tools

Source-available / open-weight tools

Commercial / hosted tools

Threat mappings

MITRE ATLAS

MAESTRO

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

Databricks AI Security Framework 3.0

AID-H-002.008: Realtime Multimodal Input Stream Security | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service