Collecter les journaux d'audit Harness IO

Compatible avec :

Ce document explique comment ingérer les journaux d'audit Harness IO dans Google Security Operations à l'aide d'Amazon S3.

Avant de commencer

  • Instance Google SecOps
  • Accès privilégié à Harness (clé API et ID de compte)
  • Accès privilégié à AWS (S3, IAM, Lambda, EventBridge)

Obtenir la clé API et l'ID de compte Harness pour un compte personnel

  1. Connectez-vous à l'interface utilisateur Web Harness.
  2. Accédez à votre profil utilisateur> Mes clés API.
  3. Sélectionnez Clé API.
  4. Saisissez un nom pour la clé API.
  5. Cliquez sur Enregistrer.
  6. Sélectionnez Jeton sous votre nouvelle clé API.
  7. Saisissez un nom pour le jeton.
  8. Cliquez sur Générer un jeton.
  9. Copiez et enregistrez le jeton dans un emplacement sécurisé.
  10. Copiez et enregistrez votre ID de compte (qui s'affiche dans l'URL Harness et dans Account Settings [Paramètres du compte]).

Facultatif : Obtenir la clé API et l'ID de compte Harness pour un compte de service

  1. Connectez-vous à l'interface utilisateur Web Harness.
  2. Créez un compte de service.
  3. Accédez à Paramètres du compte> Contrôle des accès.
  4. Sélectionnez Comptes de service> sélectionnez le compte de service pour lequel vous souhaitez créer une clé API.
  5. Sous Clés API, sélectionnez Clé API.
  6. Saisissez un nom pour la clé API.
  7. Cliquez sur Enregistrer.
  8. Sélectionnez Jeton sous la nouvelle clé API.
  9. Saisissez un nom pour le jeton.
  10. Cliquez sur Générer un jeton.
  11. Copiez et enregistrez le jeton dans un emplacement sécurisé.
  12. Copiez et enregistrez votre ID de compte (qui s'affiche dans l'URL Harness et dans Account Settings [Paramètres du compte]).

Configurer un bucket AWS S3 et IAM pour Google SecOps

  1. Créez un bucket Amazon S3 en suivant ce guide de l'utilisateur : Créer un bucket.
  2. Enregistrez le nom et la région du bucket pour référence ultérieure (par exemple, harness-io).
  3. Créez un utilisateur en suivant ce guide : Créer un utilisateur IAM.
  4. Sélectionnez l'utilisateur créé.
  5. Sélectionnez l'onglet Informations d'identification de sécurité.
  6. Cliquez sur Créer une clé d'accès dans la section Clés d'accès.
  7. Sélectionnez Service tiers comme Cas d'utilisation.
  8. Cliquez sur Suivant.
  9. Facultatif : ajoutez un tag de description.
  10. Cliquez sur Créer une clé d'accès.
  11. Cliquez sur Télécharger le fichier CSV pour enregistrer la clé d'accès et la clé d'accès secrète pour une utilisation ultérieure.
  12. Cliquez sur OK.
  13. Sélectionnez l'onglet Autorisations.
  14. Cliquez sur Ajouter des autorisations dans la section Règles d'autorisation.
  15. Sélectionnez Ajouter des autorisations.
  16. Sélectionnez Joindre directement des règles.
  17. Recherchez et sélectionnez la règle AmazonS3FullAccess.
  18. Cliquez sur Suivant.
  19. Cliquez sur Ajouter des autorisations.

Configurer la stratégie et le rôle IAM pour les importations S3

  1. Dans la console AWS, accédez à IAM > Policies > Create policy > onglet JSON.

  2. Saisissez la règle suivante :

    {
  "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"
    }
  ]
}
    • Remplacez harness-io si vous avez saisi un autre nom de bucket :

  3. Cliquez sur Suivant > Créer une règle.

  4. Accédez à IAM > Rôles > Créer un rôle > Service AWS > Lambda.

  5. Associez la règle que vous venez de créer.

  6. Nommez le rôle WriteHarnessToS3Role, puis cliquez sur Créer un rôle.

Créer la fonction Lambda

  1. Dans la console AWS, accédez à Lambda > Fonctions > Créer une fonction.
  2. Cliquez sur Créer à partir de zéro.

  3. Fournissez les informations de configuration suivantes :

    Paramètre Valeur
    Nom harness_io_to_s3
    Durée d'exécution Python 3.13
    Architecture x86_64
    Rôle d'exécution WriteHarnessToS3Role

  4. Une fois la fonction créée, ouvrez l'onglet Code, supprimez le stub et saisissez le code suivant (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. Accédez à Configuration> Variables d'environnement> Modifier> Ajouter une variable d'environnement.

  6. Saisissez les variables d'environnement suivantes en remplaçant les valeurs par les vôtres :

    Clé Exemple
    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. Une fois la fonction créée, restez sur sa page (ou ouvrez Lambda > Fonctions > votre‑fonction).

  8. Accédez à l'onglet Configuration.

  9. Dans le panneau Configuration générale, cliquez sur Modifier.

  10. Définissez le délai avant expiration sur 5 minutes (300 secondes), puis cliquez sur Enregistrer.

Créer une programmation EventBridge

  1. Accédez à Amazon EventBridge> Scheduler> Create schedule (Créer une programmation).

  2. Fournissez les informations de configuration suivantes :

    • Planning récurrent : Tarif (1 hour).
    • Cible : votre fonction Lambda.
    • Nom : harness-io-1h.

  3. Cliquez sur Créer la programmation.

Créer un utilisateur et des clés IAM en lecture seule pour Google SecOps

  1. Dans la console AWS, accédez à IAM> Utilisateurs, puis cliquez sur Ajouter des utilisateurs.
  2. Fournissez les informations de configuration suivantes :
    • Utilisateur : saisissez un nom unique (par exemple, secops-reader).
    • Type d'accès : sélectionnez Clé d'accès – Accès programmatique.
    • Cliquez sur Créer un utilisateur.
  3. Associez une règle de lecture minimale (personnalisée) : Utilisateurs> sélectionnez secops-reader > Autorisations> Ajouter des autorisations> Associer des règles directement> Créer une règle

  4. Dans l'éditeur JSON, saisissez la stratégie suivante :

    {
  "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. Définissez le nom sur secops-reader-policy.

  6. Accédez à Créer une règle > recherchez/sélectionnez > Suivant > Ajouter des autorisations.

  7. Accédez à Identifiants de sécurité > Clés d'accès > Créer une clé d'accès.

  8. Téléchargez le CSV (ces valeurs sont saisies dans le flux).

Configurer un flux dans Google SecOps pour ingérer les journaux Harness IO

  1. Accédez à Paramètres SIEM> Flux.
  2. Cliquez sur Add New Feed (Ajouter un flux).
  3. Dans le champ Nom du flux, saisissez un nom pour le flux (par exemple, Harness IO).
  4. Sélectionnez Amazon S3 V2 comme type de source.
  5. Sélectionnez Harness IO comme type de journal.
  6. Cliquez sur Suivant.
  7. Spécifiez les valeurs des paramètres d'entrée suivants :
    • URI S3 : s3://harness-io/harness/audit/
    • Options de suppression de la source : sélectionnez l'option de suppression de votre choix.
    • Âge maximal des fichiers : 180 jours par défaut.
    • ID de clé d'accès : clé d'accès utilisateur ayant accès au bucket S3.
    • Clé d'accès secrète : clé secrète de l'utilisateur ayant accès au bucket S3.
    • Espace de noms de l'élément : espace de noms de l'élément.
    • Libellés d'ingestion : libellé à appliquer aux événements de ce flux.
  8. Cliquez sur Suivant.
  9. Vérifiez la configuration de votre nouveau flux sur l'écran Finaliser, puis cliquez sur Envoyer.

Vous avez encore besoin d'aide ? Obtenez des réponses de membres de la communauté et de professionnels Google SecOps.