Recolha registos de auditoria da Harness IO

Compatível com:

Este documento explica como carregar registos de auditoria do Harness IO para o Google Security Operations através do Amazon S3.

Antes de começar

  • Instância do Google SecOps
  • Acesso privilegiado ao Harness (ID da conta e chave da API)
  • Acesso privilegiado à AWS (S3, IAM, Lambda, EventBridge)

Obtenha a chave da API e o ID da conta do Harness para uma conta pessoal

  1. Inicie sessão na IU Web do Harness.
  2. Aceda ao seu perfil de utilizador > As minhas chaves da API.
  3. Selecione Chave da API.
  4. Introduza um Nome para a chave da API.
  5. Clique em Guardar.
  6. Selecione Token na nova chave da API.
  7. Introduza um Nome para o token.
  8. Clique em Gerar token.
  9. Copie e guarde o token num local seguro.
  10. Copie e guarde o ID da conta (apresentado no URL do Harness e nas Definições da conta).

Opcional: obtenha a chave da API e o ID da conta do Harness para uma conta de serviço

  1. Inicie sessão na IU Web do Harness.
  2. Crie uma conta de serviço
  3. Aceda a Account Settings > Access Control (Definições da conta > Controlo de acesso).
  4. Selecione Contas de serviço > selecione a conta de serviço para a qual quer criar uma chave da API.
  5. Em Chaves da API, selecione Chave da API.
  6. Introduza um Nome para a chave da API.
  7. Clique em Guardar.
  8. Selecione Token na nova chave da API.
  9. Introduza um nome para o token.
  10. Clique em Gerar token.
  11. Copie e guarde o token num local seguro.
  12. Copie e guarde o ID da conta (apresentado no URL do Harness e nas Definições da conta).

Configure o contentor do AWS S3 e o IAM para o Google SecOps

  1. Crie um contentor do Amazon S3 seguindo este manual do utilizador: Criar um contentor
  2. Guarde o nome e a região do contentor para referência futura (por exemplo, harness-io).
  3. Crie um utilizador seguindo este guia do utilizador: Criar um utilizador do IAM.
  4. Selecione o utilizador criado.
  5. Selecione o separador Credenciais de segurança.
  6. Clique em Criar chave de acesso na secção Chaves de acesso.
  7. Selecione Serviço de terceiros como o Exemplo de utilização.
  8. Clicar em Seguinte.
  9. Opcional: adicione uma etiqueta de descrição.
  10. Clique em Criar chave de acesso.
  11. Clique em Transferir ficheiro CSV para guardar a chave de acesso e a chave de acesso secreta para utilização posterior.
  12. Clique em Concluído.
  13. Selecione o separador Autorizações.
  14. Clique em Adicionar autorizações na secção Políticas de autorizações.
  15. Selecione Adicionar autorizações.
  16. Selecione Anexar políticas diretamente
  17. Pesquise e selecione a política AmazonS3FullAccess.
  18. Clicar em Seguinte.
  19. Clique em Adicionar autorizações.

Configure a política e a função de IAM para carregamentos do S3

  1. Na consola da AWS, aceda a IAM > Políticas > Criar política > separador JSON.

  2. Introduza a seguinte política:

    {
  "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"
    }
  ]
}
    • Substitua harness-io se tiver introduzido um nome de contentor diferente):

  3. Clique em Seguinte > Criar política.

  4. Aceda a IAM > Funções > Criar função > Serviço AWS > Lambda.

  5. Anexe a política criada recentemente.

  6. Dê o nome WriteHarnessToS3Role à função e clique em Criar função.

Crie a função Lambda

  1. Na consola da AWS, aceda a Lambda > Functions > Create function.
  2. Clique em Criar do zero.

  3. Faculte os seguintes detalhes de configuração:

    Definição Valor
    Nome harness_io_to_s3
    Runtime Python 3.13
    Arquitetura x86_64
    Função de execução WriteHarnessToS3Role

  4. Depois de criar a função, abra o separador Código, elimine o fragmento de código e introduza o seguinte código (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. Aceda a Configuração > Variáveis de ambiente > Editar > Adicionar nova variável de ambiente.

  6. Introduza as seguintes variáveis de ambiente, substituindo-as pelos seus valores:

    Chave Exemplo
    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. Depois de criar a função, permaneça na respetiva página (ou abra Lambda > Functions > your‑function).

  8. Selecione o separador Configuração.

  9. No painel Configuração geral, clique em Editar.

  10. Altere Tempo limite para 5 minutos (300 segundos) e clique em Guardar.

Crie um horário do EventBridge

  1. Aceda a Amazon EventBridge > Scheduler > Create schedule.

  2. Faculte os seguintes detalhes de configuração:

    • Agenda recorrente: Taxa (1 hour).
    • Alvo: a sua função Lambda.
    • Nome: harness-io-1h.

  3. Clique em Criar programação.

Crie um utilizador e chaves da IAM só de leitura para o Google SecOps

  1. Na consola da AWS, aceda a IAM > Users e, de seguida, clique em Add users.
  2. Indique os seguintes detalhes de configuração:
    • Utilizador: introduza um nome único (por exemplo, secops-reader)
    • Tipo de acesso: selecione Chave de acesso – Acesso programático
    • Clique em Criar utilizador.
  3. Anexe a política de leitura mínima (personalizada): Utilizadores > selecione secops-reader > Autorizações > Adicionar autorizações > Anexar políticas diretamente > Criar política

  4. No editor JSON, introduza a seguinte política:

    {
  "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. Defina o nome como secops-reader-policy.

  6. Aceda a Criar política > pesquise/selecione > Seguinte > Adicionar autorizações.

  7. Aceda a Credenciais de segurança > Chaves de acesso > Criar chave de acesso.

  8. Transfira o CSV (estes valores são introduzidos no feed).

Configure um feed no Google SecOps para carregar registos do Harness IO

  1. Aceda a Definições do SIEM > Feeds.
  2. Clique em Adicionar novo feed.
  3. No campo Nome do feed, introduza um nome para o feed (por exemplo, Harness IO).
  4. Selecione Amazon S3 V2 como o Tipo de origem.
  5. Selecione Harness IO como o Tipo de registo.
  6. Clicar em Seguinte.
  7. Especifique valores para os seguintes parâmetros de entrada:
    • URI do S3: s3://harness-io/harness/audit/
    • Opções de eliminação de origens: selecione a opção de eliminação de acordo com a sua preferência.
    • Idade máxima do ficheiro: predefinição de 180 dias.
    • ID da chave de acesso: chave de acesso do utilizador com acesso ao contentor do S3.
    • Chave de acesso secreta: chave secreta do utilizador com acesso ao contentor do S3.
    • Espaço de nomes do recurso: o espaço de nomes do recurso.
    • Etiquetas de carregamento: a etiqueta a aplicar aos eventos deste feed.
  8. Clicar em Seguinte.
  9. Reveja a nova configuração do feed no ecrã Finalizar e, de seguida, clique em Enviar.

Precisa de mais ajuda? Receba respostas de membros da comunidade e profissionais da Google SecOps.