trailofbits-buttercup/orchestrator/docs/telemetry/README.md

CRS Telemetry Specification v1.0

For Telemetry Best Practices, see here

Telemetry generated by a CRS is submitted in two distinct ways:

1. GenAI (LLM requests and responses)
2. CRS Actions

GenAI (LLM requests and responses)

GenAI Semantic Convention is being used to track LLM Telemetry, including Prompts and Responses.

Learn more here: https://opentelemetry.io/docs/specs/semconv/gen-ai/

Any OpenTelemetry-compatible library may be used to generate and submit this LLM telemetry data. If teams are using Python or TypeScript, OpenLIT is known to work well. See OpenLIT documentation: https://docs.openlit.io/latest/quickstart-observability

import openlit
openlit.init()

If you are are not using OpenLIT, and instead using the native OpenTelemetry Instrumentation Library, make sure OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=true

CRS Actions

In order to track Tasks that a CRS is working on, the metadata field of each TaskDetail (received by the CRS API) should be added to each Span as attributes. The metadata is a string,string map. For example:

for key, value in task_data["metadata"].items():
  span.set_attribute(key, value)

Resource Attributes

These attributes describe the overarching context of the telemetry data.

Attribute Key	Type	Description
service.name	string	Name of CRS component generating the span

Mandatory Span Attributes

These attributes capture the details of individual tasks or activities.

Attribute Key	Type	Description
crs.action.category	string	Category of the work a CRS is performing.
		Values: static_analysis, dynamic_analysis, fuzzing, program_analysis, building, input_generation, patch_generation, testing, scoring_submission.
crs.action.name	string	Descriptive name of a CRS action. User-defined
<Task metadata>		Comes from Task request given to the CRS API. This should be passed on as attributes.

Optional Attributes

Attribute Key	Type	Description
crs.action.code.file	string	Code file involved in the task.
crs.action.code.lines	string	Line numbers involved in the task, if applicable.
crs.action.target.harness	string	Identifier for the target harness involved.
crs.action.effort.weight	integer	Measure of effort, e.g., number of processes or cores used.

Specialized Attributes for LLM Requests

Attribute Key	Type	Description
crs.action.gen_ai.id	string	GenAI Response ID tied to the CRS action.

Specialized Attributes for Fuzzing

Attribute Key	Type	Description
fuzz.corpus.update.method	string	Update method for the input corpus: periodic, on_change, etc.
fuzz.corpus.update.time	datetime	Timestamp of the corpus update.
fuzz.corpus.size	integer	Size of the input corpus at the time of logging.
fuzz.corpus.additions	array	List of new inputs added to the corpus.
fuzz.corpus.full_snapshot	boolean	Indicates whether this is a full snapshot of the corpus.

---

Example Python Implementation

Change the endpoint and authentication to your telemetry solution as appropriate.

1. Install OpenLIT

pip install openlit

1. Run with environment variables set.

export OPENAI_API_KEY=<YOUR OPENAI API KEY>.
OTEL_EXPORTER_OTLP_ENDPOINT="http://127.0.0.1:4317" OTEL_EXPORTER_OTLP_HEADERS="Authorization=Basic <base64 encoded http basic credentials>" OTEL_EXPORTER_OTLP_PROTOCOL=grpc python3 main.py

import logging
from openai import OpenAI
from opentelemetry import trace
import openlit

# Initialize openlit
openlit.init(application_name="mycrsllmservice")

# Acquire a tracer
tracer = trace.get_tracer(__name__)


def get_current_task():
    # Context for the task
    return {
        "metadata": {
            "round.id": "round-1",
            "task.id": "task-1",
            "team.id": "team-1",
        }
    }


def main():
    client = OpenAI()
    task_data = get_current_task()

    with tracer.start_as_current_span("patch_generation") as span:
        logging.info("Generating a patch for the following code...")
        span.set_attribute("crs.action.category", "patch_generation")
        span.set_attribute("crs.action.name", "generating_patch_from_llm")

        for key, value in task_data["metadata"].items():
            span.set_attribute(key, value)

        response = fetch_llm_completion(client, "Write a short poem about OpenTelemetry", "gpt-4o-mini")

        span.set_attribute("crs.action.gen_ai.id", response.id)

        print(response)


def fetch_llm_completion(client, prompt, model):
    response = client.chat.completions.create(
        model=model,
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": prompt,
                    }
                ],
            },
        ],
        response_format={"type": "text"},
        temperature=1,
        max_completion_tokens=2048,
        top_p=1,
        frequency_penalty=0,
        presence_penalty=0,
    )

    return response


if __name__ == "__main__":
    main()

Alternative Python Implementation

import logging
from openai import OpenAI
from opentelemetry import trace
import openlit

# Initialize openlit
openlit.init(application_name="mycrsllmservice")

# Acquire a tracer
tracer = trace.get_tracer(__name__)

def get_current_task():
    # Context for the task
    return {
        "metadata": {
            "round.id": "round-1",
            "task.id": "task-1",
            "team.id": "team-1",
        }
    }


def log_action(crs_action_category: str, crs_action_name: str, task_metadata: dict, extra_attributes: dict | None = None):
    extra_attributes = extra_attributes or {}
    with tracer.start_as_current_span(crs_action_category) as span:
        span.set_attribute("crs.action.category", crs_action_category)
        span.set_attribute("crs.action.name", crs_action_name)

        for key, value in task_metadata.items():
            span.set_attribute(key, value)

        for key, value in extra_attributes.items():
            span.set_attribute(key, value)

        span.set_status(Status(StatusCode.OK))

    print(f"Logged crs action: {crs_action_category} - {crs_action_name}")

def main():
  client = OpenAI()
  task_metadata = get_current_task()["metadata"]

  response = fetch_llm_completion(client, "Write a short poem about OpenTelemetry", "gpt-4o-mini")

  log_action("static_analysis", "check_security_flaws", task_metadata, {"crs.action.gen_ai.id": response.id})


def fetch_llm_completion(client, prompt, model):
    response = client.chat.completions.create(
        model=model,
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": prompt,
                    }
                ],
            },
        ],
        response_format={"type": "text"},
        temperature=1,
        max_completion_tokens=2048,
        top_p=1,
        frequency_penalty=0,
        presence_penalty=0,
    )

    return response


if __name__ == "__main__":
    main()

Example CRS Webservice

CRS web service example adapted from example-crs-webservice/my_crs/task_server/server.py

@app.post(
    "/v1/task/",
    response_model=None,
    responses={"202": {"model": str}},
    tags=["task"],
)
def post_v1_task_(
    credentials: Annotated[HTTPBasicCredentials, Depends(check_auth)],
    body: Task,
) -> Optional[str]:
    """
    Receive Task
    """
    tasks: List[TaskDetail] = body.tasks
    for task in tasks:
        do_task(task)

def do_task(task: TaskDetail):
    metadata_dict: dict = task.metadata
    log_action("category", "name", metadata_dict, {})

def log_action(crs_action_category: str, crs_action_name: str, task_metadata: dict, extra_attributes: dict | None = None):
    extra_attributes = extra_attributes or {}
    with tracer.start_as_current_span(crs_action_category) as span:
        span.set_attribute("crs.action.category", crs_action_category)
        span.set_attribute("crs.action.name", crs_action_name)

        for key, value in task_metadata.items():
            span.set_attribute(key, value)

        for key, value in extra_attributes.items():
            span.set_attribute(key, value)

        span.set_status(Status(StatusCode.OK))

    print(f"Logged crs action: {crs_action_category} - {crs_action_name}")

---

Other Usage Examples

Static Analysis

{
  "crs.action.category": "static_analysis",
  "crs.action.name": "analyze_vulnerabilities",
  "crs.action.code.file": "file1.c",
  "crs.action.code.lines": "10-50"
}

Dynamic Analysis

{
  "crs.action.category": "dynamic_analysis",
  "crs.action.name": "runtime_behavior_monitoring",
  "crs.action.code.file": "main.c",
  "crs.action.code.lines": "120-180",
  "crs.action.target.harness": "test_harness_1",
  "crs.action.effort.weight": 8
}

Fuzzing

{
  "crs.action.category": "fuzzing",
  "crs.action.name": "fuzz_test_network_inputs",
  "crs.action.target.harness": "network_harness",
  "fuzz.corpus.update.method": "periodic",
  "fuzz.corpus.size": 1500,
  "fuzz.corpus.additions": ["inputA", "inputB"]
}

Program Analysis

{
  "crs.action.category": "program_analysis",
  "crs.action.name": "analyze_control_flow",
  "crs.action.code.file": "utils.c"
}

Building

{
  "crs.action.category": "building",
  "crs.action.name": "compile_source_code",
  "crs.action.code.file": "module1.c"
}

Input Generation

{
  "crs.action.category": "input_generation",
  "crs.action.name": "generate_test_cases",
  "crs.action.code.file": "input_parser.c"
}

Patch Generation

{
  "crs.action.category": "patch_generation",
  "crs.action.name": "generate_security_patch",
  "crs.action.code.files": "vuln_module.c"
}

Testing

{
  "crs.action.category": "testing",
  "crs.action.name": "run_unit_tests"
}

Scoring Submission

{
  "crs.action.category": "scoring_submission",
  "crs.action.name": "submit_final_results"
}

---