Mengumpulkan log audit Harness IO

Didukung di:

Dokumen ini menjelaskan cara menyerap log audit Harness IO ke Google Security Operations menggunakan Amazon S3.

Sebelum memulai

Pastikan Anda memiliki prasyarat berikut:

  • Instance Google SecOps
  • Akses istimewa ke Harness dengan izin untuk:
    • Membuat kunci API
    • Mengakses log audit
    • Melihat setelan akun
  • Akses istimewa ke AWS (S3, IAM, Lambda, EventBridge).

Mengumpulkan kredensial Harness API

Membuat kunci API di Harness

  1. Login ke Harness Platform.
  2. Klik Profil Pengguna Anda.
  3. Buka Kunci API Saya.
  4. Klik + API Key.
  5. Berikan detail konfigurasi berikut:
    • Name: Masukkan nama deskriptif (misalnya, Google SecOps Integration).
    • Deskripsi: Deskripsi opsional.
  6. Klik Simpan.
  7. Klik + Token untuk membuat token baru.
  8. Berikan detail konfigurasi berikut:
    • Nama: Masukkan Chronicle Feed Token.
    • Setel Masa Aktif: Pilih waktu habis masa berlaku yang sesuai atau Tidak Ada Masa Aktif (untuk penggunaan produksi).
  9. Klik Generate Token.

  10. Salin dan simpan nilai token dengan aman. Token ini akan digunakan sebagai nilai header x-api-key.

Mendapatkan ID Akun Harness

  1. Di Harness Platform, catat Account ID dari URL.
    • Contoh URL: https://app.harness.io/ng/account/YOUR_ACCOUNT_ID/...
    • Bagian YOUR_ACCOUNT_ID adalah ID Akun Anda.
  2. Atau, buka Setelan Akun > Ringkasan untuk melihat ID Akun Anda.

  3. Salin dan simpan ID Akun untuk digunakan dalam fungsi Lambda.

Mengonfigurasi bucket AWS S3 dan IAM untuk Google SecOps

  1. Buat bucket Amazon S3 dengan mengikuti panduan pengguna ini: Membuat bucket
  2. Simpan Name dan Region bucket untuk referensi di masa mendatang (misalnya, harness-io-logs).
  3. Buat Pengguna dengan mengikuti panduan pengguna ini: Membuat pengguna IAM.
  4. Pilih Pengguna yang dibuat.
  5. Pilih tab Kredensial keamanan.
  6. Klik Create Access Key di bagian Access Keys.
  7. Pilih Layanan pihak ketiga sebagai Kasus penggunaan.
  8. Klik Berikutnya.
  9. Opsional: Tambahkan tag deskripsi.
  10. Klik Create access key.
  11. Klik Download CSV file untuk menyimpan Access Key dan Secret Access Key untuk referensi di masa mendatang.
  12. Klik Done.
  13. Pilih tab Izin.
  14. Klik Tambahkan izin di bagian Kebijakan izin.
  15. Pilih Tambahkan izin.
  16. Pilih Lampirkan kebijakan secara langsung.
  17. Cari kebijakan AmazonS3FullAccess.
  18. Pilih kebijakan.
  19. Klik Berikutnya.
  20. Klik Add permissions.

Mengonfigurasi kebijakan dan peran IAM untuk upload S3 Lambda

  1. Di konsol AWS, buka IAM > Policies > Create policy > JSON tab.

  2. Salin dan tempel kebijakan berikut:

    {
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowPutHarnessObjects",
      "Effect": "Allow",
      "Action": "s3:PutObject",
      "Resource": "arn:aws:s3:::harness-io-logs/harness/audit/*"
    },
    {
      "Sid": "AllowGetStateObject",
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::harness-io-logs/harness/audit/state.json"
    }
  ]
}
    • Ganti harness-io-logs jika Anda memasukkan nama bucket yang berbeda.

  3. Klik Berikutnya.

  4. Beri nama kebijakan HarnessToS3Policy, lalu klik Buat kebijakan.

  5. Buka IAM > Roles > Create role.

  6. Pilih Layanan AWS sebagai jenis entitas tepercaya.

  7. Pilih Lambda sebagai kasus penggunaan.

  8. Klik Berikutnya.

  9. Cari dan pilih kebijakan berikut:

    • HarnessToS3Policy (kebijakan yang baru saja Anda buat)
    • AWSLambdaBasicExecutionRole (untuk CloudWatch Logs)

  10. Klik Berikutnya.

  11. Beri nama peran HarnessAuditLambdaRole, lalu klik Buat peran.

Buat fungsi Lambda

  1. Di AWS Console, buka Lambda > Functions > Create function.
  2. Klik Buat dari awal.

  3. Berikan detail konfigurasi berikut:

    Setelan Nilai
    Nama harness-audit-to-s3
    Runtime Python 3.13
    Arsitektur x86_64
    Peran eksekusi HarnessAuditLambdaRole

  4. Klik Create function.

  5. Setelah fungsi dibuat, buka tab Code.

  6. Hapus kode stub default dan masukkan kode fungsi Lambda berikut:

    • Kode fungsi Lambda (harness_audit_to_s3.py)

      #!/usr/bin/env python3
"""
Harness.io Audit Logs to S3 Lambda
Fetches audit logs from Harness API and writes to S3 for Chronicle ingestion.
"""
import os
import json
import time
import uuid
import logging
import urllib.parse
from datetime import datetime, timedelta, timezone
from urllib.request import Request, urlopen
from urllib.error import HTTPError, URLError
import boto3

# Configuration from Environment Variables
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"]
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", "50")), 100)
START_MINUTES_BACK = int(os.environ.get("START_MINUTES_BACK", "60"))

# Optional filters (NEW)
FILTER_MODULES = os.environ.get("FILTER_MODULES", "").split(",") if os.environ.get("FILTER_MODULES") else None
FILTER_ACTIONS = os.environ.get("FILTER_ACTIONS", "").split(",") if os.environ.get("FILTER_ACTIONS") else None
STATIC_FILTER = os.environ.get("STATIC_FILTER")  # e.g., "EXCLUDE_LOGIN_EVENTS"
MAX_RETRIES = int(os.environ.get("MAX_RETRIES", "3"))

# AWS clients
s3 = boto3.client("s3")

# HTTP headers for Harness API
HDRS = {
    "x-api-key": API_KEY,
    "Content-Type": "application/json",
    "Accept": "application/json",
}

# Logging configuration
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# ============================================
# State Management Functions
# ============================================
def _read_state():
    """Read checkpoint state from S3."""
    try:
        obj = s3.get_object(Bucket=BUCKET, Key=STATE_KEY)
        state = json.loads(obj["Body"].read())

        since_ms = state.get("since")
        page_token = state.get("pageToken")

        logger.info(f"State loaded: since={since_ms}, pageToken={page_token}")
        return since_ms, page_token

    except s3.exceptions.NoSuchKey:
        logger.info("No state file found, starting fresh collection")
        start_time = datetime.now(timezone.utc) - timedelta(minutes=START_MINUTES_BACK)
        since_ms = int(start_time.timestamp() * 1000)
        logger.info(f"Initial since timestamp: {since_ms} ({start_time.isoformat()})")
        return since_ms, None

    except Exception as e:
        logger.error(f"Error reading state: {e}")
        raise

def _write_state(since_ms: int, page_token: str = None):
    """Write checkpoint state to S3."""
    state = {
        "since": since_ms,
        "pageToken": page_token,
        "lastRun": int(time.time() * 1000),
        "lastRunISO": datetime.now(timezone.utc).isoformat()
    }

    try:
        s3.put_object(
            Bucket=BUCKET,
            Key=STATE_KEY,
            Body=json.dumps(state, indent=2).encode(),
            ContentType="application/json"
        )
        logger.info(f"State saved: since={since_ms}, pageToken={page_token}")
    except Exception as e:
        logger.error(f"Error writing state: {e}")
        raise

# ============================================
# Harness API Functions
# ============================================
def _fetch_harness_audits(since_ms: int, page_token: str = None, retry_count: int = 0):
    """
    Fetch audit logs from Harness API with retry logic.

    API Endpoint: POST /audit/api/audits/listV2
    Documentation: https://apidocs.harness.io/audit/getauditeventlistv2
    """
    try:
        # Build URL with query parameters
        url = (
            f"{API_BASE}/audit/api/audits/listV2"
            f"?accountIdentifier={urllib.parse.quote(ACCOUNT_ID)}"
            f"&pageSize={PAGE_SIZE}"
        )

        if page_token:
            url += f"&pageToken={urllib.parse.quote(page_token)}"

        logger.info(f"Fetching from: {url[:100]}...")

        # Build request body with time filter and optional filters
        body_data = {
            "startTime": since_ms,
            "endTime": int(time.time() * 1000),
            "filterType": "Audit" 
        }

        if FILTER_MODULES:
            body_data["modules"] = [m.strip() for m in FILTER_MODULES if m.strip()]
            logger.info(f"Applying module filter: {body_data['modules']}")

        if FILTER_ACTIONS:
            body_data["actions"] = [a.strip() for a in FILTER_ACTIONS if a.strip()]
            logger.info(f"Applying action filter: {body_data['actions']}")

        if STATIC_FILTER:
            body_data["staticFilter"] = STATIC_FILTER
            logger.info(f"Applying static filter: {STATIC_FILTER}")

        logger.debug(f"Request body: {json.dumps(body_data)}")

        # Make POST request
        req = Request(
            url,
            data=json.dumps(body_data).encode('utf-8'),
            headers=HDRS,
            method="POST"
        )

        resp = urlopen(req, timeout=30)
        resp_text = resp.read().decode('utf-8')
        resp_data = json.loads(resp_text)

        if "status" not in resp_data:
            logger.warning(f"Response missing 'status' field: {resp_text[:200]}")

        # Check response status
        if resp_data.get("status") != "SUCCESS":
            error_msg = resp_data.get("message", "Unknown error")
            raise Exception(f"API returned status: {resp_data.get('status')} - {error_msg}")

        # Extract data from response structure
        data_obj = resp_data.get("data", {})

        if not data_obj:
            logger.warning("Response 'data' object is empty or missing")

        events = data_obj.get("content", [])
        has_next = data_obj.get("hasNext", False)
        next_token = data_obj.get("pageToken")

        logger.info(f"API response: {len(events)} events, hasNext={has_next}, pageToken={next_token}")

        if not events and data_obj:
            logger.info(f"Empty events but data present. Data keys: {list(data_obj.keys())}")

        return {
            "events": events,
            "hasNext": has_next,
            "pageToken": next_token
        }

    except HTTPError as e:
        error_body = e.read().decode() if hasattr(e, 'read') else ''

        if e.code == 401:
            logger.error("Authentication failed: Invalid API key")
            raise Exception("Invalid Harness API key. Check HARNESS_API_KEY environment variable.")

        elif e.code == 403:
            logger.error("Authorization failed: Insufficient permissions")
            raise Exception("API key lacks required audit:read permissions")

        elif e.code == 429:
            retry_after = int(e.headers.get("Retry-After", "60"))
            logger.warning(f"Rate limit exceeded. Retry after {retry_after} seconds (attempt {retry_count + 1}/{MAX_RETRIES})")

            if retry_count < MAX_RETRIES:
                logger.info(f"Waiting {retry_after} seconds before retry...")
                time.sleep(retry_after)
                logger.info(f"Retrying request (attempt {retry_count + 2}/{MAX_RETRIES})")
                return _fetch_harness_audits(since_ms, page_token, retry_count + 1)
            else:
                raise Exception(f"Max retries ({MAX_RETRIES}) exceeded for rate limiting")

        elif e.code == 400:
            logger.error(f"Bad request: {error_body}")
            raise Exception(f"Invalid request parameters: {error_body}")

        else:
            logger.error(f"HTTP {e.code}: {e.reason} - {error_body}")
            raise Exception(f"Harness API error {e.code}: {e.reason}")

    except URLError as e:
        logger.error(f"Network error: {e.reason}")
        raise Exception(f"Failed to connect to Harness API: {e.reason}")

    except json.JSONDecodeError as e:
        logger.error(f"Invalid JSON response: {e}")
        logger.error(f"Response text (first 500 chars): {resp_text[:500] if 'resp_text' in locals() else 'N/A'}")
        raise Exception("Harness API returned invalid JSON")

    except Exception as e:
        logger.error(f"Unexpected error in _fetch_harness_audits: {e}", exc_info=True)
        raise

# ============================================
# S3 Upload Functions
# ============================================
def _upload_to_s3(events: list) -> str:
    """
    Upload audit events to S3 in JSONL format.
    Each line is a complete JSON object (one event per line).
    """
    if not events:
        logger.info("No events to upload")
        return None

    try:
        # Create JSONL content (one JSON object per line)
        jsonl_lines = [json.dumps(event) for event in events]
        jsonl_content = "\n".join(jsonl_lines)

        # Generate S3 key with timestamp and UUID
        timestamp = datetime.now(timezone.utc)
        key = (
            f"{PREFIX}/"
            f"{timestamp:%Y/%m/%d}/"
            f"harness-audit-{timestamp:%Y%m%d-%H%M%S}-{uuid.uuid4()}.jsonl"
        )

        # Upload to S3
        s3.put_object(
            Bucket=BUCKET,
            Key=key,
            Body=jsonl_content.encode('utf-8'),
            ContentType="application/x-ndjson",
            Metadata={
                "event-count": str(len(events)),
                "source": "harness-audit-lambda",
                "collection-time": timestamp.isoformat()
            }
        )

        logger.info(f"Uploaded {len(events)} events to s3://{BUCKET}/{key}")
        return key

    except Exception as e:
        logger.error(f"Error uploading to S3: {e}", exc_info=True)
        raise

# ============================================
# Main Orchestration Function
# ============================================
def fetch_and_store():
    """
    Main function to fetch audit logs from Harness and store in S3.
    Handles pagination and state management.
    """
    logger.info("=== Harness Audit Collection Started ===")
    logger.info(f"Configuration: API_BASE={API_BASE}, ACCOUNT_ID={ACCOUNT_ID[:8]}..., PAGE_SIZE={PAGE_SIZE}")

    if FILTER_MODULES:
        logger.info(f"Module filter enabled: {FILTER_MODULES}")
    if FILTER_ACTIONS:
        logger.info(f"Action filter enabled: {FILTER_ACTIONS}")
    if STATIC_FILTER:
        logger.info(f"Static filter enabled: {STATIC_FILTER}")

    try:
        # Step 1: Read checkpoint state
        since_ms, page_token = _read_state()

        if page_token:
            logger.info(f"Resuming pagination from saved pageToken")
        else:
            since_dt = datetime.fromtimestamp(since_ms / 1000, tz=timezone.utc)
            logger.info(f"Starting new collection from: {since_dt.isoformat()}")

        # Step 2: Collect all events with pagination
        all_events = []
        current_page_token = page_token
        page_count = 0
        max_pages = 100  # Safety limit
        has_next = True

        while has_next and page_count < max_pages:
            page_count += 1
            logger.info(f"--- Fetching page {page_count} ---")

            # Fetch one page of results
            result = _fetch_harness_audits(since_ms, current_page_token)

            # Extract events
            events = result.get("events", [])
            all_events.extend(events)

            logger.info(f"Page {page_count}: {len(events)} events (total: {len(all_events)})")

            # Check pagination status
            has_next = result.get("hasNext", False)
            current_page_token = result.get("pageToken")

            if not has_next:
                logger.info("Pagination complete (hasNext=False)")
                break

            if not current_page_token:
                logger.warning("hasNext=True but no pageToken, stopping pagination")
                break

            # Small delay between pages to avoid rate limiting
            time.sleep(0.5)

        if page_count >= max_pages:
            logger.warning(f"Reached max pages limit ({max_pages}), stopping")

        # Step 3: Upload collected events to S3
        if all_events:
            s3_key = _upload_to_s3(all_events)
            logger.info(f"Successfully uploaded {len(all_events)} total events")
        else:
            logger.info("No new events to upload")
            s3_key = None

        # Step 4: Update checkpoint state
        if not has_next:
            # Pagination complete - update since to current time for next run
            new_since = int(time.time() * 1000)
            _write_state(new_since, None)
            logger.info(f"Pagination complete, state updated with new since={new_since}")
        else:
            # Pagination incomplete - save pageToken for continuation
            _write_state(since_ms, current_page_token)
            logger.info(f"Pagination incomplete, saved pageToken for next run")

        # Step 5: Return result
        result = {
            "statusCode": 200,
            "message": "Success",
            "eventsCollected": len(all_events),
            "pagesProcessed": page_count,
            "paginationComplete": not has_next,
            "s3Key": s3_key,
            "filters": {
                "modules": FILTER_MODULES,
                "actions": FILTER_ACTIONS,
                "staticFilter": STATIC_FILTER
            }
        }

        logger.info(f"Collection completed: {json.dumps(result)}")
        return result

    except Exception as e:
        logger.error(f"Collection failed: {e}", exc_info=True)

        result = {
            "statusCode": 500,
            "message": "Error",
            "error": str(e),
            "errorType": type(e).__name__
        }

        return result

    finally:
        logger.info("=== Harness Audit Collection Finished ===")

# ============================================
# Lambda Handler
# ============================================
def lambda_handler(event, context):
    """AWS Lambda handler function."""
    return fetch_and_store()

# ============================================
# Local Testing
# ============================================
if __name__ == "__main__":
    # For local testing
    result = lambda_handler(None, None)
    print(json.dumps(result, indent=2))

  7. Klik Deploy untuk menyimpan kode fungsi.

Mengonfigurasi variabel lingkungan Lambda

  1. Di halaman fungsi Lambda, pilih tab Configuration.
  2. Klik Environment variables di sidebar kiri.
  3. Klik Edit.

  4. Klik Add environment variable untuk setiap variabel berikut:

    Variabel lingkungan yang diperlukan:

    Kunci Nilai Deskripsi
    HARNESS_ACCOUNT_ID ID Akun Harness Anda ID akun dari Harness
    HARNESS_API_KEY Token kunci API Anda Token dengan izin audit:read
    S3_BUCKET harness-io-logs Nama bucket S3
    S3_PREFIX harness/audit Awalan untuk objek S3
    STATE_KEY harness/audit/state.json Jalur file status di S3

    Variabel lingkungan opsional:

    Kunci Nilai Default Deskripsi
    HARNESS_API_BASE https://app.harness.io URL dasar Harness API
    PAGE_SIZE 50 Peristiwa per halaman (maks. 100)
    START_MINUTES_BACK 60 Periode lihat kembali awal dalam menit
    FILTER_MODULES Tidak ada Modul yang dipisahkan koma (misalnya, CD,CI,CE)
    FILTER_ACTIONS Tidak ada Tindakan yang dipisahkan koma (misalnya, CREATE,UPDATE,DELETE)
    STATIC_FILTER Tidak ada Filter standar: EXCLUDE_LOGIN_EVENTS atau EXCLUDE_SYSTEM_EVENTS
    MAX_RETRIES 3 Upaya percobaan ulang maksimum untuk pembatasan kapasitas

  5. Klik Simpan.

Mengonfigurasi waktu tunggu dan memori Lambda

  1. Di halaman fungsi Lambda, pilih tab Configuration.
  2. Klik Konfigurasi umum di sidebar kiri.
  3. Klik Edit.
  4. Berikan detail konfigurasi berikut:
    • Memori: 256 MB (direkomendasikan)
    • Waktu tunggu: 5 min 0 sec (300 detik)
  5. Klik Simpan.

Membuat jadwal EventBridge

  1. Buka Amazon EventBridge > Scheduler > Create schedule.
  2. Berikan detail konfigurasi berikut:
    • Nama jadwal: Masukkan harness-audit-hourly.
    • Deskripsi: Deskripsi opsional.
  3. Klik Berikutnya.
  4. Di bagian Pola jadwal, pilih Jadwal berulang.
  5. Pilih Jadwal berbasis tarif.
  6. Berikan detail konfigurasi berikut:
    • Ekspresi tarif: Masukkan 1 hour.
  7. Klik Berikutnya.
  8. Di bagian Target, berikan detail konfigurasi berikut:
    • Target API: Pilih AWS Lambda Invoke.
    • Fungsi Lambda: Pilih fungsi Anda harness-audit-to-s3.
  9. Klik Berikutnya.
  10. Tinjau konfigurasi jadwal.
  11. Klik Buat jadwal.

Membuat pengguna IAM hanya baca untuk Google SecOps

Pengguna IAM ini memungkinkan Google SecOps membaca log dari bucket S3.

  1. Buka Konsol AWS > IAM > Users > Create user.
  2. Berikan detail konfigurasi berikut:
    • Nama pengguna: Masukkan chronicle-s3-reader.
  3. Klik Berikutnya.
  4. Pilih Lampirkan kebijakan secara langsung.
  5. Klik Create policy.
  6. Pilih tab JSON.

  7. Tempelkan kebijakan berikut:

    {
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "s3:GetObject"
      ],
      "Resource": "arn:aws:s3:::harness-io-logs/harness/audit/*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "s3:ListBucket"
      ],
      "Resource": "arn:aws:s3:::harness-io-logs",
      "Condition": {
        "StringLike": {
          "s3:prefix": "harness/audit/*"
        }
      }
    }
  ]
}

  8. Klik Berikutnya.

  9. Beri nama kebijakan ChronicleHarnessS3ReadPolicy.

  10. Klik Create policy.

  11. Kembali ke tab pembuatan pengguna dan muat ulang daftar kebijakan.

  12. Telusuri dan pilih ChronicleHarnessS3ReadPolicy.

  13. Klik Berikutnya.

  14. Tinjau, lalu klik Buat pengguna.

Membuat kunci akses untuk pengguna pembaca

  1. Di halaman IAM Users, pilih pengguna chronicle-s3-reader.
  2. Pilih tab Kredensial keamanan.
  3. Klik Create access key.
  4. Pilih Layanan pihak ketiga sebagai kasus penggunaan.
  5. Klik Berikutnya.
  6. Opsional: Tambahkan tag deskripsi.
  7. Klik Create access key.
  8. Klik Download CSV file untuk menyimpan ID Kunci Akses dan Kunci Akses Rahasia.
  9. Klik Done.

Mengonfigurasi feed di Google SecOps untuk menyerap log Harness IO

  1. Buka Setelan SIEM > Feed.
  2. Klik Tambahkan baru.
  3. Di halaman berikutnya, klik Konfigurasi satu feed.
  4. Di kolom Nama feed, masukkan nama untuk feed (misalnya, Harness Audit Logs).
  5. Pilih Amazon S3 V2 sebagai Jenis sumber.
  6. Pilih Harness IO sebagai Log type.
  7. Klik Berikutnya.

  8. Tentukan nilai untuk parameter input berikut:

    • URI S3: Masukkan URI bucket S3 dengan jalur awalan: s3://harness-io-logs/harness/audit/

    • Opsi penghapusan sumber: Pilih opsi penghapusan sesuai preferensi Anda:

      • Jangan pernah: Tidak pernah menghapus file apa pun setelah transfer (direkomendasikan pada awalnya).
      • Saat berhasil: Menghapus semua file dan direktori kosong setelah transfer berhasil.

    • Usia File Maksimum: Menyertakan file yang diubah dalam jumlah hari terakhir. Default-nya adalah 180 hari.

    • Access Key ID: Masukkan Access Key ID dari pengguna chronicle-s3-reader.

    • Secret Access Key: Masukkan Kunci Akses Rahasia dari pengguna chronicle-s3-reader.

    • Namespace aset: Namespace aset. Masukkan harness.audit.

    • Label penyerapan: Label opsional yang akan diterapkan ke peristiwa dari feed ini.

  9. Klik Berikutnya.

  10. Tinjau konfigurasi feed baru Anda di layar Selesaikan, lalu klik Kirim.

Perlu bantuan lain? Dapatkan jawaban dari anggota Komunitas dan profesional Google SecOps.