Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -334,13 +334,28 @@ class SpanAttributes:


class Events(Enum):
"""Standardized OpenTelemetry events emitted by AI instrumentation.

Security threat events provide a vendor-neutral representation for
detections from scanners and policy engines, improving interoperability
of threat telemetry across observability backends.
"""

DB_QUERY_EMBEDDINGS = "db.query.embeddings"
DB_QUERY_RESULT = "db.query.result"
DB_SEARCH_EMBEDDINGS = "db.search.embeddings"
DB_SEARCH_RESULT = "db.search.result"
GEN_AI_SECURITY_THREAT_DETECTED = "gen_ai.security.threat.detected"


class EventAttributes(Enum):
"""Attributes carried by standardized OpenTelemetry span events.

The security threat attributes make detections from different AI-agent
scanners interoperable by using one stable ``gen_ai.security.threat``
namespace.
"""

# Query Embeddings
DB_QUERY_EMBEDDINGS_VECTOR = "db.query.embeddings.vector"

Expand All @@ -361,6 +376,31 @@ class EventAttributes(Enum):
DB_SEARCH_RESULT_DISTANCE = "db.search.result.distance"
DB_SEARCH_RESULT_ENTITY = "db.search.result.entity"

# AI agent security threat detection
GEN_AI_SECURITY_THREAT_RULE_ID = "gen_ai.security.threat.rule_id"
GEN_AI_SECURITY_THREAT_CATEGORY = "gen_ai.security.threat.category"
GEN_AI_SECURITY_THREAT_SEVERITY = "gen_ai.security.threat.severity"
GEN_AI_SECURITY_THREAT_SCANNER_NAME = "gen_ai.security.threat.scanner_name"
GEN_AI_SECURITY_THREAT_SCANNER_VERSION = "gen_ai.security.threat.scanner_version"
GEN_AI_SECURITY_THREAT_ACTION = "gen_ai.security.threat.action"


class GenAISecurityThreatSeverityValues(Enum):
"""Severity values for ``gen_ai.security.threat.severity``."""

LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"


class GenAISecurityThreatActionValues(Enum):
"""Action values for ``gen_ai.security.threat.action``."""

BLOCKED = "blocked"
WARNED = "warned"
LOGGED = "logged"


class LLMRequestTypeValues(Enum):
COMPLETION = "completion"
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,85 @@
import re

import pytest

from opentelemetry.semconv_ai import (
EventAttributes,
Events,
GenAISecurityThreatActionValues,
GenAISecurityThreatSeverityValues,
)


def test_security_threat_event_value():
"""Verify the standardized security threat event name."""

assert Events.GEN_AI_SECURITY_THREAT_DETECTED.value == (
"gen_ai.security.threat.detected"
)


@pytest.mark.parametrize(
"attribute, expected",
[
(
EventAttributes.GEN_AI_SECURITY_THREAT_RULE_ID,
"gen_ai.security.threat.rule_id",
),
(
EventAttributes.GEN_AI_SECURITY_THREAT_CATEGORY,
"gen_ai.security.threat.category",
),
(
EventAttributes.GEN_AI_SECURITY_THREAT_SEVERITY,
"gen_ai.security.threat.severity",
),
(
EventAttributes.GEN_AI_SECURITY_THREAT_SCANNER_NAME,
"gen_ai.security.threat.scanner_name",
),
(
EventAttributes.GEN_AI_SECURITY_THREAT_SCANNER_VERSION,
"gen_ai.security.threat.scanner_version",
),
(
EventAttributes.GEN_AI_SECURITY_THREAT_ACTION,
"gen_ai.security.threat.action",
),
],
)
def test_security_threat_attribute_values(attribute, expected):
"""Verify each security threat attribute uses its exact semantic key."""

assert attribute.value == expected


def test_security_threat_names_use_dot_notation():
"""Verify threat event and attribute keys follow OTel dot notation."""

values = [
Events.GEN_AI_SECURITY_THREAT_DETECTED.value,
*(attribute.value for attribute in EventAttributes if attribute.name.startswith("GEN_AI_SECURITY_THREAT_")),
]
assert all(re.fullmatch(r"[a-z0-9_]+(?:\.[a-z0-9_]+)+", value) for value in values)
assert all(value.startswith("gen_ai.security.threat.") for value in values)


def test_security_threat_severity_values():
"""Verify the supported security threat severity values."""

assert {value.value for value in GenAISecurityThreatSeverityValues} == {
"low",
"medium",
"high",
"critical",
}


def test_security_threat_action_values():
"""Verify the supported security threat response action values."""

assert {value.value for value in GenAISecurityThreatActionValues} == {
"blocked",
"warned",
"logged",
}