Raccogliere gli audit log di Harness IO

Supportato in:

Questo documento spiega come importare i log di controllo di Harness IO in Google Security Operations utilizzando Amazon S3.

Prima di iniziare

  • Istanza Google SecOps
  • Accesso con privilegi a Harness (chiave API e ID account)
  • Accesso privilegiato ad AWS (S3, IAM, Lambda, EventBridge)

Recuperare la chiave API e l'ID account di Harness per un account personale

  1. Accedi all'interfaccia utente web di Harness.
  2. Vai al tuo profilo utente > Le mie chiavi API.
  3. Seleziona Chiave API.
  4. Inserisci un nome per la chiave API.
  5. Fai clic su Salva.
  6. Seleziona Token nella nuova chiave API.
  7. Inserisci un nome per il token.
  8. Fai clic su Genera token.
  9. Copia e salva il token in un luogo sicuro.
  10. Copia e salva il tuo ID account (visualizzato nell'URL di Harness e in Impostazioni account).

(Facoltativo) Ottieni la chiave API e l'ID account Harness per un account di servizio

  1. Accedi all'interfaccia utente web di Harness.
  2. Crea un service account.
  3. Vai a Impostazioni account > Controllo accessi.
  4. Seleziona Service account > seleziona il account di servizio per cui vuoi creare una chiave API.
  5. In Chiavi API, seleziona Chiave API.
  6. Inserisci un nome per la chiave API.
  7. Fai clic su Salva.
  8. Seleziona Token nella nuova chiave API.
  9. Inserisci un nome per il token.
  10. Fai clic su Genera token.
  11. Copia e salva il token in un luogo sicuro.
  12. Copia e salva il tuo ID account (visualizzato nell'URL di Harness e in Impostazioni account).

Configura il bucket AWS S3 e IAM per Google SecOps

  1. Crea un bucket Amazon S3 seguendo questa guida utente: Creazione di un bucket
  2. Salva il nome e la regione del bucket per riferimento futuro (ad esempio, harness-io).
  3. Crea un utente seguendo questa guida: Creazione di un utente IAM.
  4. Seleziona l'utente creato.
  5. Seleziona la scheda Credenziali di sicurezza.
  6. Fai clic su Crea chiave di accesso nella sezione Chiavi di accesso.
  7. Seleziona Servizio di terze parti come Caso d'uso.
  8. Fai clic su Avanti.
  9. (Facoltativo) Aggiungi un tag di descrizione.
  10. Fai clic su Crea chiave di accesso.
  11. Fai clic su Scarica file CSV per salvare la chiave di accesso e la chiave di accesso segreta per un utilizzo successivo.
  12. Fai clic su Fine.
  13. Seleziona la scheda Autorizzazioni.
  14. Fai clic su Aggiungi autorizzazioni nella sezione Criteri per le autorizzazioni.
  15. Seleziona Aggiungi autorizzazioni.
  16. Seleziona Allega direttamente i criteri.
  17. Cerca e seleziona il criterio AmazonS3FullAccess.
  18. Fai clic su Avanti.
  19. Fai clic su Aggiungi autorizzazioni.

Configura il ruolo e il criterio IAM per i caricamenti S3

  1. Nella console AWS, vai a IAM > Policy > Crea policy > Scheda JSON.

  2. Inserisci la seguente policy:

    {
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowPutHarnessObjects",
      "Effect": "Allow",
      "Action": "s3:PutObject",
      "Resource": "arn:aws:s3:::harness-io/*"
    },
    {
      "Sid": "AllowGetStateObject",
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::harness-io/harness/audit/state.json"
    }
  ]
}
    • Sostituisci harness-io se hai inserito un nome bucket diverso:

  3. Fai clic su Avanti > Crea policy.

  4. Vai a IAM > Ruoli > Crea ruolo > Servizio AWS > Lambda.

  5. Allega il criterio appena creato.

  6. Assegna al ruolo il nome WriteHarnessToS3Role e fai clic su Crea ruolo.

Crea la funzione Lambda

  1. Nella console AWS, vai a Lambda > Funzioni > Crea funzione.
  2. Fai clic su Crea autore da zero.

  3. Fornisci i seguenti dettagli di configurazione:

    Impostazione Valore
    Nome harness_io_to_s3
    Tempo di esecuzione Python 3.13
    Architettura x86_64
    Ruolo di esecuzione WriteHarnessToS3Role

  4. Dopo aver creato la funzione, apri la scheda Codice, elimina lo stub e inserisci il seguente codice (harness_io_to_s3.py).

    #!/usr/bin/env python3

import os, json, time, urllib.parse
from urllib.request import Request, urlopen
from urllib.error import HTTPError, URLError
import boto3

API_BASE = os.environ.get("HARNESS_API_BASE", "https://app.harness.io").rstrip("/")
ACCOUNT_ID = os.environ["HARNESS_ACCOUNT_ID"]
API_KEY = os.environ["HARNESS_API_KEY"]  # x-api-key token
BUCKET = os.environ["S3_BUCKET"]
PREFIX = os.environ.get("S3_PREFIX", "harness/audit/").strip("/")
STATE_KEY = os.environ.get("STATE_KEY", "harness/audit/state.json")
PAGE_SIZE = min(int(os.environ.get("PAGE_SIZE", "100")), 100)  # <=100
START_MINUTES_BACK = int(os.environ.get("START_MINUTES_BACK", "60"))

s3 = boto3.client("s3")
HDRS = {"x-api-key": API_KEY, "Content-Type": "application/json", "Accept": "application/json"}

def _read_state():
    try:
        obj = s3.get_object(Bucket=BUCKET, Key=STATE_KEY)
        j = json.loads(obj["Body"].read())
        return j.get("since"), j.get("pageToken")
    except Exception:
        return None, None

def _write_state(since_ms: int, page_token: str | None):
    body = json.dumps({"since": since_ms, "pageToken": page_token}).encode("utf-8")
    s3.put_object(Bucket=BUCKET, Key=STATE_KEY, Body=body, ContentType="application/json")

def _http_post(path: str, body: dict, query: dict, timeout: int = 60, max_retries: int = 5) -> dict:
    qs = urllib.parse.urlencode(query)
    url = f"{API_BASE}{path}?{qs}"
    data = json.dumps(body).encode("utf-8")

    attempt, backoff = 0, 1.0
    while True:
        req = Request(url, data=data, method="POST")
        for k, v in HDRS.items():
            req.add_header(k, v)
        try:
            with urlopen(req, timeout=timeout) as r:
                return json.loads(r.read().decode("utf-8"))
        except HTTPError as e:
            if (e.code == 429 or 500 <= e.code <= 599) and attempt < max_retries:
                time.sleep(backoff)
                attempt += 1
                backoff *= 2
                continue
            raise
        except URLError:
            if attempt < max_retries:
                time.sleep(backoff)
                attempt += 1
                backoff *= 2
                continue
            raise

def _write_page(obj: dict, now: float, page_index: int) -> str:
    ts = time.strftime("%Y/%m/%d/%H%M%S", time.gmtime(now))
    key = f"{PREFIX}/{ts}-page{page_index:05d}.json"
    s3.put_object(
        Bucket=BUCKET,
        Key=key,
        Body=json.dumps(obj, separators=(",", ":")).encode("utf-8"),
        ContentType="application/json",
    )
    return key

def fetch_and_store():
    now_s = time.time()
    since_ms, page_token = _read_state()
    if since_ms is None:
        since_ms = int((now_s - START_MINUTES_BACK * 60) * 1000)
    until_ms = int(now_s * 1000)

    page_index = 0
    total = 0

    while True:
        body = {"startTime": since_ms, "endTime": until_ms}
        query = {"accountIdentifier": ACCOUNT_ID, "pageSize": PAGE_SIZE}
        if page_token:
            query["pageToken"] = page_token
        else:
            query["pageIndex"] = page_index

        data = _http_post("/audit/api/audits/listV2", body, query)
        _write_page(data, now_s, page_index)

        entries = []
        for key in ("data", "content", "response", "resource", "resources", "items"):
            if isinstance(data.get(key), list):
                entries = data[key]
                break
        total += len(entries) if isinstance(entries, list) else 0

        next_token = (
            data.get("pageToken")
            or (isinstance(data.get("meta"), dict) and data["meta"].get("pageToken"))
            or (isinstance(data.get("metadata"), dict) and data["metadata"].get("pageToken"))
        )

        if next_token:
            page_token = next_token
            page_index += 1
            continue

        if len(entries) < PAGE_SIZE:
            break
        page_index += 1

    _write_state(until_ms, None)
    return {"pages": page_index + 1, "objects_estimate": total}

def lambda_handler(event=None, context=None):
    return fetch_and_store()

if __name__ == "__main__":
    print(lambda_handler())

  5. Vai a Configurazione > Variabili di ambiente > Modifica > Aggiungi nuova variabile di ambiente.

  6. Inserisci le seguenti variabili di ambiente, sostituendole con i tuoi valori:

    Chiave Esempio
    S3_BUCKET harness-io
    S3_PREFIX harness/audit/
    STATE_KEY harness/audit/state.json
    HARNESS_ACCOUNT_ID 123456789
    HARNESS_API_KEY harness_xxx_token
    HARNESS_API_BASE https://app.harness.io
    PAGE_SIZE 100
    START_MINUTES_BACK 60

  7. Dopo aver creato la funzione, rimani sulla relativa pagina (o apri Lambda > Funzioni > la tua funzione).

  8. Seleziona la scheda Configurazione.

  9. Nel riquadro Configurazione generale, fai clic su Modifica.

  10. Modifica Timeout impostandolo su 5 minuti (300 secondi) e fai clic su Salva.

Creare una pianificazione EventBridge

  1. Vai a Amazon EventBridge > Scheduler > Crea pianificazione.

  2. Fornisci i seguenti dettagli di configurazione:

    • Programma ricorrente: Tariffa (1 hour).
    • Destinazione: la tua funzione Lambda.
    • Nome: harness-io-1h

  3. Fai clic su Crea pianificazione.

Crea un utente IAM con autorizzazione di sola lettura e chiavi per Google SecOps

  1. Nella console AWS, vai a IAM > Utenti, poi fai clic su Aggiungi utenti.
  2. Fornisci i seguenti dettagli di configurazione:
    • Utente: inserisci un nome univoco (ad esempio secops-reader)
    • Tipo di accesso: seleziona Chiave di accesso - Accesso programmatico
    • Fai clic su Crea utente.
  3. Allega criterio per la lettura minimi (personalizzati): Utenti > seleziona secops-reader > Autorizzazioni > Aggiungi autorizzazioni > Allega criteri direttamente > Crea criteri

  4. Nell'editor JSON, inserisci la seguente policy:

    {
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["s3:GetObject"],
      "Resource": "arn:aws:s3:::<your-bucket>/*"
    },
    {
      "Effect": "Allow",
      "Action": ["s3:ListBucket"],
      "Resource": "arn:aws:s3:::<your-bucket>"
    }
  ]
}

  5. Imposta il nome su secops-reader-policy.

  6. Vai a Crea criterio > cerca/seleziona > Avanti > Aggiungi autorizzazioni.

  7. Vai a Credenziali di sicurezza > Chiavi di accesso > Crea chiave di accesso.

  8. Scarica il file CSV (questi valori vengono inseriti nel feed).

Configura un feed in Google SecOps per importare i log di Harness IO

  1. Vai a Impostazioni SIEM > Feed.
  2. Fai clic su Aggiungi nuovo feed.
  3. Nel campo Nome feed, inserisci un nome per il feed (ad esempio, Harness IO).
  4. Seleziona Amazon S3 V2 come Tipo di origine.
  5. Seleziona Harness IO come Tipo di log.
  6. Fai clic su Avanti.
  7. Specifica i valori per i seguenti parametri di input:
    • URI S3: s3://harness-io/harness/audit/
    • Opzioni di eliminazione dell'origine: seleziona l'opzione di eliminazione in base alle tue preferenze.
    • Durata massima del file: 180 giorni per impostazione predefinita.
    • ID chiave di accesso: chiave di accesso utente con accesso al bucket S3.
    • Chiave di accesso segreta: chiave segreta dell'utente con accesso al bucket S3.
    • Spazio dei nomi dell'asset: lo spazio dei nomi dell'asset.
    • Etichette di importazione: l'etichetta da applicare agli eventi di questo feed.
  8. Fai clic su Avanti.
  9. Controlla la nuova configurazione del feed nella schermata Finalizza e poi fai clic su Invia.

Hai bisogno di ulteriore assistenza? Ricevi risposte dai membri della community e dai professionisti di Google SecOps.