We've reorganized our documentation navigation structure to align directly with your operational workflows. See the release notes and the walkthrough video for more information.

Collect Akamai SIEM Connector logs

Parser Version: 2.0

Supported in:

Google secops SIEM

This guide explains how you can ingest Akamai SIEM Connector logs to Google Security Operations. Akamai SIEM Integration provides security events from the Akamai platform in near real time through the SIEM Integration API in JSON format. These events include data from Web Application Firewall (WAF), Bot Manager, and other Akamai security products. Because the Akamai SIEM Integration API is pull-based, a scheduled function periodically fetches events and writes them to object storage, where Google SecOps ingests them. You can configure ingestion using one of two supported methods:

Method 1: Cloud Storage. A Cloud Run function periodically pulls events from the Akamai SIEM Integration API and writes them to a Cloud Storage bucket, where Google SecOps ingests them.
Method 2: Amazon S3. An AWS Lambda function periodically pulls events from the Akamai SIEM Integration API and writes them to an Amazon S3 bucket, where Google SecOps ingests them.

Before you begin

Ensure that you have the following prerequisites:

Common to both methods:

A Google SecOps instance.
Privileged access to Akamai Control Center with the Manage SIEM user role.

Method 1 additional prerequisites (Cloud Storage):

A Google Cloud project with the Cloud Storage, Cloud Run, Pub/Sub, and Cloud Scheduler APIs enabled.
Permissions to create and manage Cloud Storage buckets, service accounts, Cloud Run functions, Pub/Sub topics, and Cloud Scheduler jobs.

Method 2 additional prerequisites (Amazon S3):

Privileged access to AWS (S3, Identity and Access Management, Lambda, EventBridge).

Enable SIEM Integration in Akamai Control Center

Sign in to Akamai Control Center.
Go to Security Configurations under the Cloud Security section.
Open the security configuration (and the appropriate version) for which you want to collect SIEM data.
Click Advanced Settings and expand Data collection for SIEM Integrations.
Go to Security Configurations under the Cloud Security section.
Open the security configuration (and the appropriate version) for which you want to collect SIEM data.
Click Advanced Settings and expand Data collection for SIEM Integrations.
Click On to enable SIEM.
Choose the security policies for which you want to export data:
- All Security policies: Send SIEM data for events that violate any or all security policies within the security configuration.
- Specific security policies: Select one or more specific security policies from the drop-down list.
Optional: If you use Account Protector and want to include the unencrypted Username, turn on the Include username checkbox.

Note: When you include username, anyone with access to your SIEM output can potentially see this value and its associated risk score. Note: When you include username, anyone with access to your SIEM output can potentially see this value and its associated risk score.
Optional: If you want to receive JA4 fingerprint information in SIEM events, turn on the Include the JA4 Client TLS Fingerprint checkbox.
Optional: If you want to exclude events belonging to a specific protection type and action, click Add exception. Select the protection and the associated actions you don't want SIEM to collect.
Optional: If you want to receive JA4 fingerprint information in SIEM events, turn on the Include the JA4 Client TLS Fingerprint checkbox.
Optional: If you want to exclude events belonging to a specific protection type and action, click Add exception. Select the protection and the associated actions you don't want SIEM to collect.
Click Save.
Copy and save the Security Configuration ID (configId) from the SIEM Integration section. You need this value for Lambda configuration.
Copy and save the Security Configuration ID (configId) from the SIEM Integration section. You need this value for Lambda configuration.

Create Akamai API credentials for SIEM Integration

Sign in to Akamai Control Center.
Go to ACCOUNT ADMIN > Identity & access > Users and API Clients.
Go to ACCOUNT ADMIN > Identity & access > Users and API Clients.
Click Create API client.
Click Advanced to create an advanced API client.
Click Advanced to create an advanced API client.
Provide the following configuration details:
- Name: Enter a descriptive name (for example, Google SecOps SIEM Poller).
- API service: Select SIEM and set the access level to READ-WRITE.
- Name: Enter a descriptive name (for example, SecOps SIEM Poller).
- API service: Select SIEM and set the access level to READ-WRITE.
Click Create API client.
Click Download to save the credentials as an .edgerc file, or click Copy to copy the credentials. Save the following values securely:
- Client Token
- Client Secret
- Access Token
- Host (for example, akab-xxxxxxxxxxxxxxxx-xxxxxxxxxxxxxxxx.luna.akamaiapis.net)

Method 1: Cloud Storage

Use this method to collect Akamai SIEM Connector logs into a Cloud Storage bucket. A Cloud Run function, triggered by Cloud Scheduler through Pub/Sub, periodically pulls events from the Akamai SIEM Integration API and writes them to the bucket, where Google SecOps ingests them.

Create a Cloud Storage bucket

Go to the Google Cloud console.
Select your project or create a new one.
Go to Cloud Storage > Buckets.
Click Create bucket.
Provide the following configuration details:
- Name your bucket: Enter a globally unique name (for example, akamai-siem-logs).
- Location type: Choose based on your needs (Region, Dual-region, or Multi-region).
- Location: Select the location (for example, us-central1).
- Storage class: Select Standard.
- Access control: Select Uniform.
Click Create.

Create a service account for the Cloud Run function

Go to IAM & Admin > Service Accounts.
Click Create service account.
Provide the following configuration details:
- Service account name: Enter akamai-siem-collector-sa.
- Service account description: Enter Service account for Cloud Run function to collect Akamai SIEM Connector logs.
Click Create and continue.
Grant the following roles:
- Storage Object Admin: Write events to the bucket and manage the state file.
- Cloud Run Invoker: Allow Pub/Sub to invoke the function.
Click Continue, and then click Done.

Grant the service account access to the bucket

Go to Cloud Storage > Buckets.
Click your bucket name.
Go to the Permissions tab.
Click Grant access.
Provide the following configuration details:
- Add principals: Enter the service account email (akamai-siem-collector-sa@<PROJECT_ID>.iam.gserviceaccount.com).
- Assign roles: Select Storage Object Admin.
Click Save.

Create a Pub/Sub topic

Go to Pub/Sub > Topics.
Click Create topic.
In the Topic ID field, enter akamai-siem-trigger.
Click Create.

Create the Cloud Run function to collect logs

The Cloud Run function is triggered by Pub/Sub messages from Cloud Scheduler. It signs requests with the Akamai EdgeGrid credentials, fetches events from the SIEM Integration API, and writes them to the bucket.

Go to Cloud Run.
Click Create service.
Select Function to use the inline code editor.
Provide the following configuration details:
- Service name: Enter akamai-siem-collector.
- Region: Select the region matching your bucket (for example, us-central1).
- Runtime: Select Python 3.12 or later.
In the Trigger section:
1. Click Add trigger.
2. Select Pub/Sub.
3. In Select a Pub/Sub topic, select akamai-siem-trigger.
4. Click Save.
In the Authentication section, select Require authentication.
Expand Containers, Networking, Security, and on the Security tab set Service account to akamai-siem-collector-sa.

On the Containers > Variables & Secrets tab, add the following environment variables:

Variable name	Example value
`GCS_BUCKET`	`akamai-siem-logs`
`GCS_PREFIX`	`akamai-siem/`
`STATE_KEY`	`akamai-siem/state.json`
`AKAMAI_HOST`	`akab-xxxxxxxx.luna.akamaiapis.net`
`AKAMAI_CLIENT_TOKEN`	`akab-client-token-xxxx`
`AKAMAI_CLIENT_SECRET`	`your-client-secret`
`AKAMAI_ACCESS_TOKEN`	`akab-access-token-xxxx`
`AKAMAI_CONFIG_ID`	`12345`
`LIMIT`	`10000`
`WINDOW_SECONDS`	`3600`
`HTTP_TIMEOUT`	`110`
`MAX_PAGES`	`20`

Set the Request timeout to 600 seconds and Memory to 512 MiB or higher.
Click Create. After the service is created, the inline code editor opens.

Add the function code

In the Function entry point field, enter main.

In the inline code editor, replace the contents of main.py with the following code:

import functions_framework
from google.cloud import storage
import json
import os
import time
import uuid
import requests
from akamai.edgegrid import EdgeGridAuth

storage_client = storage.Client()

GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'akamai-siem/')
STATE_KEY = os.environ.get('STATE_KEY', 'akamai-siem/state.json')
AKAMAI_HOST = os.environ.get('AKAMAI_HOST')
CLIENT_TOKEN = os.environ.get('AKAMAI_CLIENT_TOKEN')
CLIENT_SECRET = os.environ.get('AKAMAI_CLIENT_SECRET')
ACCESS_TOKEN = os.environ.get('AKAMAI_ACCESS_TOKEN')
CONFIG_ID = os.environ.get('AKAMAI_CONFIG_ID')
LIMIT = int(os.environ.get('LIMIT', '10000'))
WINDOW_SEC = int(os.environ.get('WINDOW_SECONDS', '3600'))
HTTP_TIMEOUT = int(os.environ.get('HTTP_TIMEOUT', '110'))
MAX_PAGES = int(os.environ.get('MAX_PAGES', '20'))

def _session():
    """Create a requests session that signs requests with Akamai EdgeGrid."""
    session = requests.Session()
    session.auth = EdgeGridAuth(
        client_token=CLIENT_TOKEN,
        client_secret=CLIENT_SECRET,
        access_token=ACCESS_TOKEN,
    )
    return session

def _load_state(bucket):
    """Load the saved offset from GCS."""
    try:
        blob = bucket.blob(STATE_KEY)
        if blob.exists():
            return json.loads(blob.download_as_text())
    except Exception as e:
        print(f'Warning: could not load state: {e}')
    return {}

def _save_state(bucket, state):
    """Persist the offset to GCS for the next run."""
    try:
        bucket.blob(STATE_KEY).upload_from_string(
            json.dumps(state, separators=(',', ':')),
            content_type='application/json',
        )
    except Exception as e:
        print(f'Warning: could not save state: {e}')

def _fetch(session, offset, from_ts):
    """Fetch one batch of events. Use offset when available, otherwise a time window.

    The Akamai SIEM API does not allow combining the offset parameter with from/to.
    """
    url = f'https://{AKAMAI_HOST}/siem/v1/configs/{CONFIG_ID}'
    params = {'limit': LIMIT}
    if offset:
        params['offset'] = offset
    else:
        params['from'] = int(from_ts)
    resp = session.get(
        url, params=params, timeout=HTTP_TIMEOUT,
        headers={'Accept': 'application/json'},
    )
    resp.raise_for_status()
    return resp.text

def _parse_ndjson(text):
    """Split the NDJSON response into event lines and the trailing context object.

    Each line is a JSON security event; the last line is a context object that
    contains the offset to use for the next request.
    """
    lines = [ln for ln in text.splitlines() if ln.strip()]
    if not lines:
        return [], {}
    try:
        context = json.loads(lines[-1])
    except ValueError:
        return lines, {}
    if isinstance(context, dict) and 'offset' in context:
        return lines[:-1], context
    return lines, {}

def _write_events(bucket, event_lines, page_num):
    """Write a batch of raw JSON events to GCS, one event per line."""
    ts_path = time.strftime('%Y/%m/%d', time.gmtime())
    uniq = f'{int(time.time() * 1e6)}_{uuid.uuid4().hex[:8]}'
    key = f'{GCS_PREFIX}{ts_path}/akamai_siem_p{page_num:03d}_{uniq}.json'
    bucket.blob(key).upload_from_string(
        '\n'.join(event_lines),
        content_type='application/x-ndjson',
    )
    return key

@functions_framework.cloud_event
def main(cloud_event):
    """Triggered by Pub/Sub to pull Akamai SIEM events and write them to GCS."""
    if not all([GCS_BUCKET, AKAMAI_HOST, CLIENT_TOKEN, CLIENT_SECRET,
                ACCESS_TOKEN, CONFIG_ID]):
        print('Error: missing required environment variables')
        return

    bucket = storage_client.bucket(GCS_BUCKET)
    state = _load_state(bucket)
    offset = state.get('offset')
    from_ts = float(state.get('last_to_ts') or (time.time() - WINDOW_SEC))

    session = _session()
    pages = 0
    total = 0
    try:
        while pages < MAX_PAGES:
            text = _fetch(session, offset, from_ts)
            events, context = _parse_ndjson(text)
            if events:
                _write_events(bucket, events, pages + 1)
                total += len(events)
            new_offset = context.get('offset')
            if new_offset:
                offset = new_offset
            pages += 1
            # Stop when this window is drained (fewer than LIMIT events returned).
            if not new_offset or len(events) < LIMIT:
                break

        state['offset'] = offset
        state['last_to_ts'] = time.time()
        state['last_successful_run'] = time.time()
        _save_state(bucket, state)
        print(f'Retrieved {total} events across {pages} pages')
    except Exception as e:
        print(f'Error collecting Akamai SIEM logs: {e}')
        raise

Create a requirements.txt file with the following contents:

functions-framework==3.*
google-cloud-storage==2.*
requests>=2.31.0
edgegrid-python>=1.3.1

Click Deploy and wait for the deployment to complete.

Note: Akamai stops sending the SIEM API response after two minutes. Keep HTTP_TIMEOUT under 120 seconds and schedule the function frequently (for example, every five minutes) so each run completes within the window.

Create a Cloud Scheduler job

Go to Cloud Scheduler.
Click Create job.
Provide the following configuration details:
- Name: Enter akamai-siem-collector-5min.
- Region: Select the same region as the Cloud Run function.
- Frequency: Enter */5 * * * * (every five minutes).
- Timezone: Select UTC.
- Target type: Select Pub/Sub.
- Topic: Select akamai-siem-trigger.
- Message body: Enter {}.
Click Create.
To verify the setup, click Force run on the job, and then check the Cloud Run logs and the bucket to confirm that events were written.

Configure a feed in Google SecOps to ingest from Cloud Storage

Go to SIEM Settings > Feeds.
Click Add New Feed.
On the next page, click Configure a single feed.
In the Feed name field, enter a name for the feed (for example, Akamai SIEM Connector GCS).
Select Cloud Storage V2 as the Source type.
Select Akamai SIEM Connector as the Log type.
Click Get Service Account next to the Chronicle Service Account field, and copy the service account email.
Click Next.
Specify values for the following input parameters:
- Storage Bucket URI: Enter gs://akamai-siem-logs/akamai-siem/. This URL must end with a trailing forward slash (/).
- Source deletion options: Select the deletion option according to your preference.
- Maximum File Age: Include files modified in the last number of days. Default is 180 days.
Click Next.
Review your new feed configuration in the Finalize screen, and then click Submit.

Grant the Google SecOps service account access to the bucket

Go to Cloud Storage > Buckets.
Click your bucket name.
Go to the Permissions tab.
Click Grant access.
Provide the following configuration details:
- Add principals: Paste the Google SecOps service account email from the feed setup.
- Assign roles: Select Storage Object Viewer (or Storage Object Admin if you chose a deletion option).
Click Save.

Method 2: Amazon S3

Use this method to collect Akamai SIEM Connector logs into an Amazon S3 bucket using AWS Lambda and EventBridge.

Configure AWS S3 bucket and IAM for Google SecOps

Create Amazon S3 bucket following this user guide: Creating a bucket
Save bucket Name and Region for future reference (for example, akamai-siem-logs).
Create a User following this user guide: Creating an IAM user.
Select the created User.
Select Security credentials tab.
Click Create Access Key in section Access Keys.
Select Third-party service as Use case.
Click Next.
Optional: Add a description tag.
Click Create access key.
Click Download .csv file to save the Access Key and Secret Access Key for future reference.
Click Done.
Select Permissions tab.
Click Add permissions in section Permissions policies.
Select Add permissions.
Select Attach policies directly.
Search for AmazonS3FullAccess policy.
Select the policy.
Click Next.
Click Add permissions.

Configure the IAM policy and role for S3 uploads

In the AWS console, go to IAM > Policies > Create policy > JSON tab.
Copy and paste the following policy.

Policy JSON (replace akamai-siem-logs with your bucket name):

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "AllowPutObjects",
            "Effect": "Allow",
            "Action": "s3:PutObject",
            "Resource": "arn:aws:s3:::akamai-siem-logs/*"
        },
        {
            "Sid": "AllowGetStateObject",
            "Effect": "Allow",
            "Action": "s3:GetObject",
            "Resource": "arn:aws:s3:::akamai-siem-logs/akamai-siem/state.json"
        }
    ]
}

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "AllowPutObjects",
            "Effect": "Allow",
            "Action": "s3:PutObject",
            "Resource": "arn:aws:s3:::akamai-siem-logs/*"
        },
        {
            "Sid": "AllowGetStateObject",
            "Effect": "Allow",
            "Action": "s3:GetObject",
            "Resource": "arn:aws:s3:::akamai-siem-logs/akamai-siem/state.json"
        }
    ]
}

Click Next.
Enter policy name AkamaiSIEMtoS3Policy and click Create policy.
Go to IAM > Roles > Create role.
Select AWS service.
Select Lambda as the use case.
Click Next.
Search for and select the policy AkamaiSIEMtoS3Policy.
Search for and select the policy AkamaiSIEMtoS3Policy.
Click Next.
Enter role name AkamaiSIEMtoS3Role and click Create role.

Create the Lambda function

In the AWS Console, go to Lambda > Functions > Create function.
Click Author from scratch.

Provide the following configuration details:

Setting	Value
Name	`AkamaiSIEMtoS3Function`
Runtime	Python 3.13
Architecture	x86_64
Execution role	Use an existing role
Existing role	`AkamaiSIEMtoS3Role`
Setting	Value
Name	`AkamaiSIEMtoS3Function`
Runtime	Python 3.13
Architecture	x86_64
Execution role	Use an existing role
Existing role	`AkamaiSIEMtoS3Role`

Click Create function.

After the function is created, open the Code tab, delete the stub and paste the following code:

import json
import boto3
import os
import urllib3
import hmac
import hashlib
import base64
from datetime import datetime
from urllib.parse import urlparse, urljoin
```python
import json
import boto3
import os
import urllib3
import hmac
import hashlib
import base64
from datetime import datetime
from urllib.parse import urlparse, urljoin

# Configuration from environment variables
S3_BUCKET = os.environ['S3_BUCKET']
S3_PREFIX = os.environ.get('S3_PREFIX', 'akamai-siem/')
STATE_KEY = os.environ.get('STATE_KEY', 'akamai-siem/state.json')
# Configuration from environment variables
S3_BUCKET = os.environ['S3_BUCKET']
S3_PREFIX = os.environ.get('S3_PREFIX', 'akamai-siem/')
STATE_KEY = os.environ.get('STATE_KEY', 'akamai-siem/state.json')

AKAMAI_HOST = os.environ['AKAMAI_HOST']
AKAMAI_CLIENT_TOKEN = os.environ['AKAMAI_CLIENT_TOKEN']
AKAMAI_CLIENT_SECRET = os.environ['AKAMAI_CLIENT_SECRET']
AKAMAI_ACCESS_TOKEN = os.environ['AKAMAI_ACCESS_TOKEN']
AKAMAI_CONFIG_IDS = os.environ['AKAMAI_CONFIG_IDS'].split(',')
AKAMAI_HOST = os.environ['AKAMAI_HOST']
AKAMAI_CLIENT_TOKEN = os.environ['AKAMAI_CLIENT_TOKEN']
AKAMAI_CLIENT_SECRET = os.environ['AKAMAI_CLIENT_SECRET']
AKAMAI_ACCESS_TOKEN = os.environ['AKAMAI_ACCESS_TOKEN']
AKAMAI_CONFIG_IDS = os.environ['AKAMAI_CONFIG_IDS'].split(',')

LIMIT = int(os.environ.get('LIMIT', '10000'))
LIMIT = int(os.environ.get('LIMIT', '10000'))

s3_client = boto3.client('s3')
http = urllib3.PoolManager()
s3_client = boto3.client('s3')
http = urllib3.PoolManager()

def load_state():
    """Load offset state from S3"""
    try:
        response = s3_client.get_object(Bucket=S3_BUCKET, Key=STATE_KEY)
        return json.loads(response['Body'].read().decode('utf-8'))
    except s3_client.exceptions.NoSuchKey:
        return {}
    except Exception as e:
        print(f"Error loading state: {e}")
        return {}
def load_state():
    """Load offset state from S3"""
    try:
        response = s3_client.get_object(Bucket=S3_BUCKET, Key=STATE_KEY)
        return json.loads(response['Body'].read().decode('utf-8'))
    except s3_client.exceptions.NoSuchKey:
        return {}
    except Exception as e:
        print(f"Error loading state: {e}")
        return {}

def save_state(state):
    """Save offset state to S3"""
    try:
        s3_client.put_object(
            Bucket=S3_BUCKET,
            Key=STATE_KEY,
            Body=json.dumps(state, indent=2).encode('utf-8'),
            ContentType='application/json'
        )
    except Exception as e:
        print(f"Error saving state: {e}")
def save_state(state):
    """Save offset state to S3"""
    try:
        s3_client.put_object(
            Bucket=S3_BUCKET,
            Key=STATE_KEY,
            Body=json.dumps(state, indent=2).encode('utf-8'),
            ContentType='application/json'
        )
    except Exception as e:
        print(f"Error saving state: {e}")

def make_edgegrid_auth_header(url, method='GET'):
    """Create EdgeGrid authentication header"""
    timestamp = datetime.utcnow().strftime('%Y%m%dT%H:%M:%S+0000')
    nonce = base64.b64encode(os.urandom(16)).decode('utf-8')

    parsed_url = urlparse(url)
    relative_url = parsed_url.path
    if parsed_url.query:
        relative_url += '?' + parsed_url.query

    auth_header = f'EG1-HMAC-SHA256 ' \
                 f'client_token={AKAMAI_CLIENT_TOKEN};' \
                 f'access_token={AKAMAI_ACCESS_TOKEN};' \
                 f'timestamp={timestamp};' \
                 f'nonce={nonce};'

    data_to_sign = '\t'.join([
        method,
        parsed_url.scheme,
        parsed_url.netloc,
        relative_url,
        '',  # Request body for GET
        '',  # No additional headers
    ])

    signing_key = hmac.new(
        AKAMAI_CLIENT_SECRET.encode('utf-8'),
        timestamp.encode('utf-8'),
        hashlib.sha256
    ).digest()

    auth_signature = base64.b64encode(
        hmac.new(
            signing_key,
            (data_to_sign + auth_header).encode('utf-8'),
            hashlib.sha256
        ).digest()
    ).decode('utf-8')

    return auth_header + f'signature={auth_signature}'
def make_edgegrid_auth_header(url, method='GET'):
    """Create EdgeGrid authentication header"""
    timestamp = datetime.utcnow().strftime('%Y%m%dT%H:%M:%S+0000')
    nonce = base64.b64encode(os.urandom(16)).decode('utf-8')

    parsed_url = urlparse(url)
    relative_url = parsed_url.path
    if parsed_url.query:
        relative_url += '?' + parsed_url.query

    auth_header = f'EG1-HMAC-SHA256 ' \
                 f'client_token={AKAMAI_CLIENT_TOKEN};' \
                 f'access_token={AKAMAI_ACCESS_TOKEN};' \
                 f'timestamp={timestamp};' \
                 f'nonce={nonce};'

    data_to_sign = '\t'.join([
        method,
        parsed_url.scheme,
        parsed_url.netloc,
        relative_url,
        '',  # Request body for GET
        '',  # No additional headers
    ])

    signing_key = hmac.new(
        AKAMAI_CLIENT_SECRET.encode('utf-8'),
        timestamp.encode('utf-8'),
        hashlib.sha256
    ).digest()

    auth_signature = base64.b64encode(
        hmac.new(
            signing_key,
            (data_to_sign + auth_header).encode('utf-8'),
            hashlib.sha256
        ).digest()
    ).decode('utf-8')

    return auth_header + f'signature={auth_signature}'

def fetch_akamai_events(config_id, offset=None):
    """Fetch events from Akamai SIEM API"""
    base_url = f'https://{AKAMAI_HOST}'
    endpoint = f'/siem/v1/configs/{config_id}'

    params = f'limit={LIMIT}'
    if offset:
        params += f'&offset={offset}'

    url = f'{base_url}{endpoint}?{params}'

    try:
        headers = {
            'Authorization': make_edgegrid_auth_header(url)
        }

        response = http.request('GET', url, headers=headers, timeout=120)

        if response.status != 200:
            print(f"Error response {response.status}: {response.data.decode('utf-8')}")
            return [], offset

        # Parse multi-JSON response (newline-delimited JSON)
        lines = response.data.decode('utf-8').strip().split('\n')
        events = []
        new_offset = offset

        for line in lines:
            if not line.strip():
                continue
            try:
                obj = json.loads(line)

                # Check if this is offset context (metadata object with offset)
                if 'offset' in obj and ('total' in obj or 'responseContext' in obj):
                    new_offset = obj.get('offset')
                    continue

                # This is an event
                events.append(obj)
            except json.JSONDecodeError as e:
                print(f"Warning: Failed to parse line: {e}")
                continue

        return events, new_offset

    except Exception as e:
        print(f"Error fetching events for config {config_id}: {e}")
        return [], offset
def fetch_akamai_events(config_id, offset=None):
    """Fetch events from Akamai SIEM API"""
    base_url = f'https://{AKAMAI_HOST}'
    endpoint = f'/siem/v1/configs/{config_id}'

    params = f'limit={LIMIT}'
    if offset:
        params += f'&offset={offset}'

    url = f'{base_url}{endpoint}?{params}'

    try:
        headers = {
            'Authorization': make_edgegrid_auth_header(url)
        }

        response = http.request('GET', url, headers=headers, timeout=120)

        if response.status != 200:
            print(f"Error response {response.status}: {response.data.decode('utf-8')}")
            return [], offset

        # Parse multi-JSON response (newline-delimited JSON)
        lines = response.data.decode('utf-8').strip().split('\n')
        events = []
        new_offset = offset

        for line in lines:
            if not line.strip():
                continue
            try:
                obj = json.loads(line)

                # Check if this is offset context (metadata object with offset)
                if 'offset' in obj and ('total' in obj or 'responseContext' in obj):
                    new_offset = obj.get('offset')
                    continue

                # This is an event
                events.append(obj)
            except json.JSONDecodeError as e:
                print(f"Warning: Failed to parse line: {e}")
                continue

        return events, new_offset

    except Exception as e:
        print(f"Error fetching events for config {config_id}: {e}")
        return [], offset

def lambda_handler(event, context):
    """Lambda handler - fetches Akamai events and writes to S3"""
    print(f"Starting Akamai SIEM fetch at {datetime.utcnow().isoformat()}Z")

    state = load_state()
    total_events = 0

    for config_id in AKAMAI_CONFIG_IDS:
        config_id = config_id.strip()
        if not config_id:
            continue

        print(f"Fetching events for config: {config_id}")

        current_offset = state.get(config_id)
        events, new_offset = fetch_akamai_events(config_id, current_offset)

        if events:
            print(f"Fetched {len(events)} events for config {config_id}")

            # Write events to S3 as newline-delimited JSON
            timestamp = datetime.utcnow().strftime('%Y%m%d_%H%M%S')
            s3_key = f'{S3_PREFIX}{config_id}/{timestamp}.json'

            payload = '\n'.join(json.dumps(event) for event in events)

            try:
                s3_client.put_object(
                    Bucket=S3_BUCKET,
                    Key=s3_key,
                    Body=payload.encode('utf-8'),
                    ContentType='application/json'
                )
                print(f"Wrote {len(events)} events to s3://{S3_BUCKET}/{s3_key}")

                # Update offset only after successful write
                if new_offset:
                    state[config_id] = new_offset
                    total_events += len(events)
            except Exception as e:
                print(f"Error writing to S3: {e}")
        else:
            print(f"No new events for config {config_id}")

    # Save updated state
    save_state(state)

    return {
        'statusCode': 200,
        'body': json.dumps({
            'message': f'Successfully processed {total_events} events',
            'configs_processed': len(AKAMAI_CONFIG_IDS)
        })
    }

def lambda_handler(event, context): """Lambda handler - fetches Akamai events and writes to S3""" print(f"Starting Akamai SIEM fetch at {datetime.utcnow().isoformat()}Z")

state = load_state()
total_events = 0

for config_id in AKAMAI_CONFIG_IDS:
    config_id = config_id.strip()
    if not config_id:
        continue

    print(f"Fetching events for config: {config_id}")

    current_offset = state.get(config_id)
    events, new_offset = fetch_akamai_events(config_id, current_offset)

    if events:
        print(f"Fetched {len(events)} events for config {config_id}")

        # Write events to S3 as newline-delimited JSON
        timestamp = datetime.utcnow().strftime('%Y%m%d_%H%M%S')
        s3_key = f'{S3_PREFIX}{config_id}/{timestamp}.json'

        payload = '\n'.join(json.dumps(event) for event in events)

        try:
            s3_client.put_object(
                Bucket=S3_BUCKET,
                Key=s3_key,
                Body=payload.encode('utf-8'),
                ContentType='application/json'
            )
            print(f"Wrote {len(events)} events to s3://{S3_BUCKET}/{s3_key}")

            # Update offset only after successful write
            if new_offset:
                state[config_id] = new_offset
                total_events += len(events)
        except Exception as e:
            print(f"Error writing to S3: {e}")
    else:
        print(f"No new events for config {config_id}")

# Save updated state
save_state(state)

return {
    'statusCode': 200,
    'body': json.dumps({
        'message': f'Successfully processed {total_events} events',
        'configs_processed': len(AKAMAI_CONFIG_IDS)
    })
}

Click Deploy to save the code.
Go to Configuration > Environment variables > Edit.
Go to Configuration > Environment variables > Edit.
Click Add environment variable for each of the following:

Environment variables Environment variables

Key	Example value
`S3_BUCKET`	`akamai-siem-logs`
`S3_PREFIX`	`akamai-siem/`
`STATE_KEY`	`akamai-siem/state.json`
`AKAMAI_HOST`	`akab-xxxxxxxx.luna.akamaiapis.net`
`AKAMAI_CLIENT_TOKEN`	`your-client-token`
`AKAMAI_CLIENT_SECRET`	`your-client-secret`
`AKAMAI_ACCESS_TOKEN`	`your-access-token`
`AKAMAI_CONFIG_IDS`	`12345,67890`
`LIMIT`	`10000`
Key	Example value
`S3_BUCKET`	`akamai-siem-logs`
`S3_PREFIX`	`akamai-siem/`
`STATE_KEY`	`akamai-siem/state.json`
`AKAMAI_HOST`	`akab-xxxxxxxx.luna.akamaiapis.net`
`AKAMAI_CLIENT_TOKEN`	`your-client-token`
`AKAMAI_CLIENT_SECRET`	`your-client-secret`
`AKAMAI_ACCESS_TOKEN`	`your-access-token`
`AKAMAI_CONFIG_IDS`	`12345,67890`
`LIMIT`	`10000`

Click Save.
Select the Configuration tab.
In the General configuration panel click Edit.
Change Timeout to 5 minutes (300 seconds) and click Save.
Select the Configuration tab.
In the General configuration panel click Edit.
Change Timeout to 5 minutes (300 seconds) and click Save.

Note: Timeout is a hard upper limit. If the function is still running when the timer hits 5 minutes, AWS Lambda will terminate it. You are only billed for actual run time. Note: Timeout is a hard upper limit. If the function is still running when the timer hits 5 minutes, AWS Lambda will terminate it. You are only billed for actual run time.

Create an EventBridge schedule

Go to Amazon EventBridge > Scheduler > Create schedule.
Provide the following configuration details:
- Schedule name: Enter AkamaiSIEMtoS3-5min.
- Schedule pattern: Select Recurring schedule.
- Schedule type: Select Rate-based schedule.
- Rate expression: Enter 5 and select Minutes.
- Schedule name: Enter AkamaiSIEMtoS3-5min.
- Schedule pattern: Select Recurring schedule.
- Schedule type: Select Rate-based schedule.
- Rate expression: Enter 5 and select Minutes.
Click Next.
Provide the following configuration details:
- Target: Select AWS Lambda Invoke.
- Lambda function: Select AkamaiSIEMtoS3Function.
- Target: Select AWS Lambda Invoke.
- Lambda function: Select AkamaiSIEMtoS3Function.
Click Next.
Click Next (skip optional settings).
Review and click Create schedule.

Configure a feed in Google SecOps to ingest Akamai SIEM Connector logs

Go to SIEM Settings > Feeds.
Click Add New Feed.
On the next page, click Configure a single feed.
Enter a unique name for the Feed name (for example, Akamai SIEM Connector).
Click Add New Feed.
On the next page, click Configure a single feed.
Enter a unique name for the Feed name (for example, Akamai SIEM Connector).
Select Amazon S3 V2 as the Source type.
Select Akamai SIEM Connector as the Log type.
Click Next and then click Submit.
Click Next and then click Submit.
Specify values for the following input parameters:
- S3 URI: s3://akamai-siem-logs/akamai-siem/
- Source deletion option: Select the deletion option according to your preference.
- Maximum File Age: Include files modified in the last number of days. Default is 180 days.
- Access Key ID: User access key with access to the S3 bucket.
- Secret Access Key: User secret key with access to the S3 bucket.
- Asset namespace: The asset namespace.
- Ingestion labels: The label applied to the events from this feed.
- S3 URI: s3://akamai-siem-logs/akamai-siem/
- Source deletion option: Select the deletion option according to your preference.
- Maximum File Age: Include files modified in the last number of days. Default is 180 days.
- Access Key ID: User access key with access to the S3 bucket.
- Secret Access Key: User secret key with access to the S3 bucket.
- Asset namespace: The asset namespace.
- Ingestion labels: The label applied to the events from this feed.
Note: If you select the Delete transferred files or Delete transferred files and empty directories option, make sure that you granted appropriate permissions to the service account.
Click Next and then click Submit.

UDM mapping table

Log field	UDM mapping	Logic
`accept_language`	`additional.fields`	Merged
`additional_policy_key`	`additional.fields`	Merged
`attackData_label`	`additional.fields`	Merged
`attackData_ruleActions_label`	`additional.fields`	Merged
`attackData_ruleData_label`	`additional.fields`	Merged
`attackData_ruleSelectors_label`	`additional.fields`	Merged
`attackdata_ruleMessages_label`	`additional.fields`	Merged
`attackdata_ruleTags_label`	`additional.fields`	Merged
`attackdata_ruleVersions_label`	`additional.fields`	Merged
`botData_label`	`additional.fields`	Merged
`continent_label`	`additional.fields`	Merged
`dur_label`	`additional.fields`	Merged
`etag_label`	`additional.fields`	Merged
`format_label`	`additional.fields`	Merged
`httpMessage_query_label`	`additional.fields`	Merged
`httpMessage_response_headers_label`	`additional.fields`	Merged
`identity_label`	`additional.fields`	Merged
`identityv2_label`	`additional.fields`	Merged
`mime_version_label`	`additional.fields`	Merged
`regionCode_label`	`additional.fields`	Merged
`req_id_label`	`additional.fields`	Merged
`requestHeaders_label`	`additional.fields`	Merged
`responseSegment_label`	`additional.fields`	Merged
`ruleAction_list`	`additional.fields`	Merged
`ruleData_list`	`additional.fields`	Merged
`ruleMessages_list`	`additional.fields`	Merged
`ruleSelectors_list`	`additional.fields`	Merged
`ruleTags_list`	`additional.fields`	Merged
`ruleVersions_list`	`additional.fields`	Merged
`rules_list`	`additional.fields`	Merged
`same_site`	`additional.fields`	Merged
`telemetryType_label`	`additional.fields`	Merged
`Date`	`metadata.event_timestamp`	Parsed as `EEE, dd MMM yyyy HH:mm:ss ZZZ`
`expires`	`metadata.event_timestamp`	Parsed as `EEE, dd-MMM-yyyy HH:mm:ss ZZZ`
`httpMessage.start`	`metadata.event_timestamp`	Parsed as `UNIX`
`event_type`	`metadata.event_type`	Directly mapped
`attackData.configId`	`metadata.product_log_id`	Directly mapped
`version`	`metadata.product_version`	Directly mapped
`protocol`	`network.application_protocol`	Directly mapped
`MS-ASProtocolVersion`	`network.application_protocol_version`	Directly mapped
`httpMessage.method`	`network.http.method`	Directly mapped
`httpMessage.status`	`network.http.response_code`	Renamed/mapped
`User-Agent`	`network.http.user_agent`	Directly mapped
`httpMessage.bytes`	`network.sent_bytes`	Renamed/mapped
`ASP.NET_SessionId`	`network.session_id`	Directly mapped
`httpMessage.requestId`	`network.session_id`	Directly mapped
`httpMessage.tls`	`network.tls.version`	Directly mapped
`attackData.clientIP`	`principal.asset.ip`	Merged
`attackData.clientIP`	`principal.ip`	Merged
`geo.city`	`principal.location.city`	Directly mapped
`geo.country`	`principal.location.country_or_region`	Directly mapped
`geo.asn`	`principal.location.name`	Directly mapped
`additional_content_Security_Policy`	`principal.resource.attribute.labels`	Merged
`DeviceType`	`principal.resource.name`	Directly mapped
`DeviceId`	`principal.resource.product_object_id`	Directly mapped
`cache_label`	`security_result.about.labels`	Merged
`accept_encoding_label`	`security_result.about.resource.attribute.labels`	Merged
`accept_label`	`security_result.about.resource.attribute.labels`	Merged
`connection_label`	`security_result.about.resource.attribute.labels`	Merged
`content_encoding`	`security_result.about.resource.attribute.labels`	Merged
`content_length_label`	`security_result.about.resource.attribute.labels`	Merged
`content_type_label`	`security_result.about.resource.attribute.labels`	Merged
`req_content_length_label`	`security_result.about.resource.attribute.labels`	Merged
`req_content_type_label`	`security_result.about.resource.attribute.labels`	Merged
`desc`	`security_result.description`	Directly mapped
`CP_IsMobile_label`	`security_result.detection_fields`	Merged
`X-BEServer_label`	`security_result.detection_fields`	Merged
`X-CalculatedBETarget_label`	`security_result.detection_fields`	Merged
`X-Content-Type-Options_label`	`security_result.detection_fields`	Merged
`X-DiagInfo_label`	`security_result.detection_fields`	Merged
`X-FEServer_label`	`security_result.detection_fields`	Merged
`X-MS-BackOffDuration_label`	`security_result.detection_fields`	Merged
`apiId_label`	`security_result.detection_fields`	Merged
`backend_label`	`security_result.detection_fields`	Merged
`p3p_label`	`security_result.detection_fields`	Merged
`type_label`	`security_result.detection_fields`	Merged
`vary_label`	`security_result.detection_fields`	Merged
`attackData.policyId`	`security_result.summary`	Directly mapped
`Host`	`target.asset.hostname`	Directly mapped
`httpMessage.host`	`target.asset.hostname`	Directly mapped
`path`	`target.file.full_path`	Directly mapped
`Host`	`target.hostname`	Directly mapped
`httpMessage.host`	`target.hostname`	Directly mapped
`httpMessage.port`	`target.port`	Directly mapped
`Cmd`	`target.process.command_line`	Directly mapped
`httpMessage.path`	`target.url`	Directly mapped
`User`	`target.user.userid`	Directly mapped
N/A	`metadata.event_type`	Constant: `GENERIC_EVENT`
N/A	`metadata.product_name`	Constant: `AKAMAI_SIEM_CONNECTOR`
N/A	`metadata.vendor_name`	Constant: `AKAMAI_SIEM_CONNECTOR`
N/A	`principal.resource.resource_type`	Constant: `DEVICE`

Change Log

View the Change Log for this parser

Need more help? Get answers from Community members and Google SecOps professionals.