SDK Integration & Custom Transforms

# monad_chla/transforms/ise.py

from dataclasses import dataclass
from typing import List, Dict, Any

# Message ID classifications
ISE_CRITICAL_IDS = [
    5400,   # Auth failed
    5405,   # RADIUS dropped
    5440,   # Endpoint abandoned
    5449,   # Endpoint EAP failed
    12508,  # EAP-TLS failed
    86010,  # Guest failed
]

ISE_BULK_IDS = [
    5200,   # Auth passed
    24015,  # EAP started
    24016,  # EAP continued
    86009,  # Guest login
]

@dataclass
class ISEParserConfig:
    """Configuration for ISE syslog parser transform."""
    name: str = "ise-syslog-parser"

    # Regex patterns for ISE syslog fields
    patterns: Dict[str, str] = None

    def __post_init__(self):
        self.patterns = {
            "message_id": r"(\d{4,5})\s+NOTICE",
            "username": r"UserName=([^,\s]+)",
            "mac_address": r"Calling-Station-ID=([0-9A-Fa-f:-]+)",
            "ip_address": r"Framed-IP-Address=([0-9.]+)",
            "nas_ip": r"NAS-IP-Address=([0-9.]+)",
            "nas_port": r"NAS-Port=(\d+)",
            "failure_reason": r"FailureReason=([^,]+)",
            "auth_policy": r"AuthenticationPolicy=([^,]+)",
            "authz_policy": r"AuthorizationPolicy=([^,]+)",
            "selected_profile": r"SelectedAuthorizationProfiles=([^,]+)",
        }

    def to_monad_config(self) -> Dict[str, Any]:
        """Convert to Monad API format."""
        return {
            "name": self.name,
            "type": "parse",
            "config": {
                "rules": [
                    {"field": k, "pattern": v}
                    for k, v in self.patterns.items()
                ]
            }
        }


@dataclass
class ISEFilterConfig:
    """Configuration for ISE critical/bulk filter transform."""
    name: str = "ise-critical-filter"
    critical_ids: List[int] = None

    def __post_init__(self):
        if self.critical_ids is None:
            self.critical_ids = ISE_CRITICAL_IDS

    def to_monad_config(self) -> Dict[str, Any]:
        """Convert to Monad API format."""
        return {
            "name": self.name,
            "type": "filter",
            "config": {
                "condition": f"message_id in {self.critical_ids}"
            }
        }


class ISETransformBuilder:
    """Builder for ISE transform pipeline."""

    def __init__(self, api_client, org_id: str):
        self.api = api_client
        self.org_id = org_id

    def create_parser(self, config: ISEParserConfig = None) -> str:
        """Create ISE parser transform, return transform ID."""
        config = config or ISEParserConfig()
        response = self.api.V1OrganizationIdTransformsPost(
            self.org_id,
            config.to_monad_config()
        )
        return response.id

    def create_critical_filter(self, config: ISEFilterConfig = None) -> str:
        """Create ISE critical filter transform, return transform ID."""
        config = config or ISEFilterConfig()
        response = self.api.V1OrganizationIdTransformsPost(
            self.org_id,
            config.to_monad_config()
        )
        return response.id

    def create_bulk_filter(self) -> str:
        """Create ISE bulk filter (inverse of critical)."""
        config = {
            "name": "ise-bulk-filter",
            "type": "filter",
            "config": {
                "condition": f"message_id in {ISE_BULK_IDS}"
            }
        }
        response = self.api.V1OrganizationIdTransformsPost(
            self.org_id,
            config
        )
        return response.id

# monad_chla/transforms/ftd.py

from dataclasses import dataclass
from typing import List, Dict, Any

FTD_CRITICAL_IDS = [
    "106006", "106007", "106014", "106015", "106023",
    "430003", "733100"
]

FTD_BULK_IDS = [
    "106001",  # Permitted
    "302013",  # Built connection
    "302014",  # Teardown connection
]

@dataclass
class FTDParserConfig:
    """Configuration for FTD syslog parser transform."""
    name: str = "ftd-syslog-parser"

    patterns: Dict[str, str] = None

    def __post_init__(self):
        self.patterns = {
            "syslog_id": r"%FTD-\d-(\d{6})",
            "action": r"(permitted|denied|Deny|Drop)",
            "src_ip": r"(\d+\.\d+\.\d+\.\d+)/\d+",
            "dst_ip": r"->.*?(\d+\.\d+\.\d+\.\d+)/\d+",
            "src_port": r"(\d+\.\d+\.\d+\.\d+)/(\d+)",
            "dst_port": r"->.*?(\d+\.\d+\.\d+\.\d+)/(\d+)",
            "protocol": r"(TCP|UDP|ICMP|GRE)",
            "acl_name": r"access-group\s+(\S+)",
            "interface": r"interface\s+(\S+)",
        }

    def to_monad_config(self) -> Dict[str, Any]:
        return {
            "name": self.name,
            "type": "parse",
            "config": {
                "rules": [
                    {"field": k, "pattern": v}
                    for k, v in self.patterns.items()
                ]
            }
        }


class FTDTransformBuilder:
    """Builder for FTD transform pipeline."""

    def __init__(self, api_client, org_id: str):
        self.api = api_client
        self.org_id = org_id

    def create_parser(self, config: FTDParserConfig = None) -> str:
        config = config or FTDParserConfig()
        response = self.api.V1OrganizationIdTransformsPost(
            self.org_id,
            config.to_monad_config()
        )
        return response.id

    def create_critical_filter(self) -> str:
        config = {
            "name": "ftd-critical-filter",
            "type": "filter",
            "config": {
                "condition": f"syslog_id in {FTD_CRITICAL_IDS} or action in ['denied', 'Deny', 'Drop']"
            }
        }
        response = self.api.V1OrganizationIdTransformsPost(
            self.org_id,
            config
        )
        return response.id

Week 1-2: jq fundamentals
├── Filter, select, capture
├── Reshape objects
└── Type conversions

Week 3-4: Python SDK
├── API client wrapper
├── Transform builders
└── Pipeline deployment scripts

Week 5-6: Go SDK (optional)
├── Convert Python transforms to Go
├── Performance-critical pipelines
└── Native Monad integration

# pipelines/ise-radius/config.yaml

pipeline:
  name: ise-radius-to-sentinel
  description: ISE RADIUS logs - critical to Sentinel, bulk to S3

input:
  type: s3
  config:
    bucket: chla-syslog-staging
    prefix: ise/radius/
    region: us-west-2

transforms:
  - name: ise-parser
    type: custom
    module: monad_chla.transforms.ise
    class: ISEParserConfig

  - name: ise-critical-filter
    type: custom
    module: monad_chla.transforms.ise
    class: ISEFilterConfig
    params:
      critical_ids: [5400, 5405, 5440, 5449, 12508]

outputs:
  critical:
    type: sentinel
    condition: "passes critical filter"
    config:
      workspace_id: ${SENTINEL_WORKSPACE_ID}
      shared_key: ${SENTINEL_SHARED_KEY}
      log_type: ISESecurityLogs

  archive:
    type: s3
    condition: "all logs"
    config:
      bucket: chla-log-archive
      prefix: ise/radius/
      region: us-west-2

#!/usr/bin/env python3
# scripts/deploy.py

import os
import yaml
import argparse
from monad_chla.client import MonadClient
from monad_chla.transforms.ise import ISETransformBuilder
from monad_chla.transforms.ftd import FTDTransformBuilder

def load_config(pipeline_path: str) -> dict:
    """Load pipeline configuration from YAML."""
    with open(f"pipelines/{pipeline_path}/config.yaml") as f:
        return yaml.safe_load(f)

def deploy_pipeline(client: MonadClient, config: dict) -> str:
    """Deploy a complete pipeline, return pipeline ID."""

    # Create input
    input_id = client.create_input(config["input"])
    print(f"Created input: {input_id}")

    # Create transforms
    transform_ids = []
    for t in config["transforms"]:
        if t["type"] == "custom":
            # Use our custom builders
            builder = get_transform_builder(client, t["module"])
            transform_id = builder.create_from_config(t)
        else:
            transform_id = client.create_transform(t)
        transform_ids.append(transform_id)
        print(f"Created transform: {transform_id}")

    # Create outputs
    output_ids = []
    for name, output_config in config["outputs"].items():
        output_id = client.create_output(output_config)
        output_ids.append(output_id)
        print(f"Created output ({name}): {output_id}")

    # Wire together as pipeline
    pipeline_id = client.create_pipeline(
        name=config["pipeline"]["name"],
        description=config["pipeline"]["description"],
        nodes=[input_id] + transform_ids + output_ids,
        edges=build_edges(input_id, transform_ids, output_ids, config)
    )

    print(f"Created pipeline: {pipeline_id}")
    return pipeline_id

def main():
    parser = argparse.ArgumentParser(description="Deploy Monad pipeline")
    parser.add_argument("pipeline", help="Pipeline to deploy (e.g., ise-radius)")
    parser.add_argument("--dry-run", action="store_true", help="Validate only")
    args = parser.parse_args()

    client = MonadClient(
        api_key=os.environ["MONAD_API_KEY"],
        org_id=os.environ["MONAD_ORG_ID"]
    )

    config = load_config(args.pipeline)

    if args.dry_run:
        print(f"Would deploy: {config['pipeline']['name']}")
        return

    pipeline_id = deploy_pipeline(client, config)
    print(f"Successfully deployed: {pipeline_id}")

if __name__ == "__main__":
    main()

# .github/workflows/deploy.yaml

name: Deploy Monad Pipelines

on:
  push:
    branches: [main]
    paths:
      - 'pipelines/**'
      - 'monad_chla/**'
  workflow_dispatch:
    inputs:
      pipeline:
        description: 'Pipeline to deploy'
        required: true
        type: choice
        options:
          - ise-radius
          - ise-tacacs
          - ftd-firewall
          - network-devices
          - all

jobs:
  validate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - run: pip install -e .
      - run: python scripts/validate.py

  deploy:
    needs: validate
    runs-on: ubuntu-latest
    environment: production
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - run: pip install -e .
      - name: Deploy pipeline
        env:
          MONAD_API_KEY: ${{ secrets.MONAD_API_KEY }}
          MONAD_ORG_ID: ${{ secrets.MONAD_ORG_ID }}
          SENTINEL_WORKSPACE_ID: ${{ secrets.SENTINEL_WORKSPACE_ID }}
          SENTINEL_SHARED_KEY: ${{ secrets.SENTINEL_SHARED_KEY }}
        run: |
          if [ "${{ github.event.inputs.pipeline }}" = "all" ]; then
            for p in ise-radius ise-tacacs ftd-firewall network-devices; do
              python scripts/deploy.py $p
            done
          else
            python scripts/deploy.py ${{ github.event.inputs.pipeline }}
          fi

# tests/test_ise_transforms.py

import pytest
from monad_chla.transforms.ise import ISEParserConfig, ISEFilterConfig

class TestISEParser:

    def test_parser_extracts_message_id(self):
        config = ISEParserConfig()
        # Verify regex pattern matches ISE log format
        import re
        sample = "5400 NOTICE Failed-Attempt: UserName=jsmith"
        match = re.search(config.patterns["message_id"], sample)
        assert match.group(1) == "5400"

    def test_parser_extracts_username(self):
        config = ISEParserConfig()
        import re
        sample = "UserName=john.doe, Calling-Station-ID=AA:BB:CC:DD:EE:FF"
        match = re.search(config.patterns["username"], sample)
        assert match.group(1) == "john.doe"

    def test_critical_filter_includes_failures(self):
        config = ISEFilterConfig()
        assert 5400 in config.critical_ids
        assert 5405 in config.critical_ids

    def test_critical_filter_excludes_success(self):
        config = ISEFilterConfig()
        assert 5200 not in config.critical_ids

class TestISEFilterConfig:

    def test_monad_config_format(self):
        config = ISEFilterConfig()
        monad_config = config.to_monad_config()

        assert monad_config["name"] == "ise-critical-filter"
        assert monad_config["type"] == "filter"
        assert "condition" in monad_config["config"]