Falcon API Integration in Functions

from logging import Logger
from typing import Any, Dict, Union
from crowdstrike.foundry.function import Function, Request, Response
from falconpy import Alerts, Hosts

func = Function.instance()

@func.handler(method='GET', path='/api/alerts')
def get_alerts(request: Request, config: Union[Dict[str, Any], None], logger: Logger) -> Response:
    falcon = Alerts()  # Zero-arg constructor — auth is automatic

    limit = min(int(request.params.get("limit", 50)), 100)
    # FQL filter: high-severity alerts from the last 24 hours.
    # Combine conditions with '+' (AND); relative times like 'now-24h' are supported.
    response = falcon.query_alerts_v2(
        filter="severity_name:'High'+created_timestamp:>'now-24h'",
        limit=limit,
        sort="created_timestamp|desc",
    )

    if response["status_code"] != 200:
        logger.error(f"Failed to query alerts: {response.get('errors')}")
        return Response(body={"error": "Failed to fetch alerts"}, code=500)

    alert_ids = response.get("body", {}).get("resources", [])
    if not alert_ids:
        return Response(body={"alerts": []}, code=200)

    details_response = falcon.get_alerts_v2(ids=alert_ids)
    if details_response["status_code"] != 200:
        return Response(body={"error": "Failed to fetch alert details"}, code=500)

    alerts = details_response.get("body", {}).get("resources", [])
    return Response(body={"alerts": alerts}, code=200)

if __name__ == '__main__':
    func.run()

package main

import (
    "context"
    "log/slog"
    "github.com/crowdstrike/gofalcon/falcon"
    "github.com/crowdstrike/gofalcon/falcon/client"
    fdk "github.com/crowdstrike/foundry-fn-go"
)

func newHandler(_ context.Context, _ *slog.Logger, _ fdk.SkipCfg) fdk.Handler {
    m := fdk.NewMux()

    m.Get("/api/alerts", fdk.HandleFnOf(func(ctx context.Context, r fdk.RequestOf[struct{}]) fdk.Response {
        accessToken := r.Header.Get("X-CS-ACCESSTOKEN")

        opts := fdk.FalconClientOpts()
        falconClient, err := falcon.NewClient(&falcon.ApiConfig{
            AccessToken:       accessToken,
            Cloud:             falcon.Cloud(opts.Cloud),
            Context:           ctx,
            UserAgentOverride: opts.UserAgent,
        })
        if err != nil {
            return fdk.Response{Code: 500, Body: fdk.JSON(map[string]string{"error": "Failed to authenticate"})}
        }

        // ... API calls with falconClient ...
        return fdk.Response{Code: 200, Body: fdk.JSON(map[string]interface{}{"alerts": []interface{}{}})}
    }))

    return m
}

func main() {
    fdk.Run(context.Background(), newHandler)
}

@func.handler(method='GET', path='/api/detections')
def get_detections(request: Request, config, logger) -> Response:
    falcon = Alerts()  # Zero-arg — auth is automatic

    severity_min = int(request.params.get("severity_min", 3))
    limit = min(int(request.params.get("limit", 50)), 100)

    # Use Alerts v2 methods — these hit /alerts/entities/alerts/v3 under the hood.
    # FQL filter: severity threshold + product "detections" (excludes cases/incidents).
    query_response = falcon.query_alerts_v2(
        filter=f"severity:>='{severity_min}'+product:'detections'",
        limit=limit,
        sort="created_timestamp|desc",
    )
    if query_response["status_code"] != 200:
        return Response(body={"error": "Failed to query detections"}, code=500)

    alert_ids = query_response.get("body", {}).get("resources", [])
    if not alert_ids:
        return Response(body={"detections": []}, code=200)

    details = falcon.get_alerts_v2(ids=alert_ids)
    if details["status_code"] != 200:
        return Response(body={"error": "Failed to get details"}, code=500)

    return Response(body={"detections": details["body"]["resources"]}, code=200)

@func.handler(method='GET', path='/api/hosts/{hostname}')
def get_host_details(request: Request, config, logger) -> Response:
    falcon = Hosts()

    hostname = request.params.get("hostname")
    if not hostname:
        return Response(body={"error": "Hostname required"}, code=400)

    query = falcon.query_devices_by_filter(filter=f"hostname:'{hostname}'")
    if query["status_code"] != 200:
        return Response(body={"error": "Failed to query devices"}, code=500)

    host_ids = query.get("body", {}).get("resources", [])
    if not host_ids:
        return Response(body={"error": f"Host not found: {hostname}"}, code=404)

    details = falcon.get_device_details(ids=host_ids)
    host = details.get("body", {}).get("resources", [{}])[0]
    return Response(body={"host": host}, code=200)

import time
from logging import Logger
from typing import Any, Dict, Union
from crowdstrike.foundry.function import Function, Request, Response
from falconpy import NGSIEM

func = Function.instance()

# Always "search-all" — see the gotcha below. Scope to a repo in the query, not here.
REPO = "search-all"


def run_logscale_query(ngsiem, query_string, start, end, logger, max_wait=40):
    """Execute a CQL query as an async search job and return result rows.

    ``start`` / ``end`` accept Humio relative strings ("24h", "7d", "30d",
    "now") or epoch-millisecond integers.
    """
    payload = {"queryString": query_string, "start": start, "end": end, "isLive": False}
    # Must be search=, not body= — see the keyword gotcha below.
    started = ngsiem.start_search(repository=REPO, search=payload)

    if not isinstance(started, dict) or started.get("status_code", 500) >= 300:
        logger.error(f"start_search failed: {started}")
        return None

    # start_search returns "resources"; get_search_status returns "body".
    job_id = (started.get("resources") or {}).get("id")
    if not job_id:
        logger.error(f"start_search returned no job id: {started}")
        return None

    # Poll until done
    waited = 0.0
    poll_interval = 1.5
    while waited < max_wait:
        time.sleep(poll_interval)
        waited += poll_interval
        status = ngsiem.get_search_status(repository=REPO, id=job_id)
        if not isinstance(status, dict) or status.get("status_code", 500) >= 300:
            logger.error(f"get_search_status failed: {status}")
            return None
        sbody = status.get("body") or {}
        if sbody.get("done"):
            return sbody.get("events", []) or []

    logger.warning(f"query timed out after {max_wait}s: {query_string}")
    return None


@func.handler(method='POST', path='/api/query')
def handle_query(request: Request, config: Union[Dict[str, Any], None], logger: Logger) -> Response:
    ngsiem = NGSIEM()  # Zero-arg auth — automatic in Foundry

    query = request.body.get("query", "")
    start = request.body.get("start", "24h")
    end = request.body.get("end", "now")

    events = run_logscale_query(ngsiem, query, start, end, logger)
    if events is None:
        return Response(body={"error": "LogScale query failed"}, code=500)

    return Response(body={"results": events, "count": len(events)}, code=200)

if __name__ == '__main__':
    func.run()

# Filter to a specific repo within the query itself
query = "#repo=fusion | execution_log_type=summary | execution_log_subtype=end | groupby([status], function=count(execution_id, distinct=true))"
events = run_logscale_query(ngsiem, query, "24h", "now", logger)

# manifest.yml
auth:
    scopes:
        - humio-auth-proxy:read     # Required for queries
        - humio-auth-proxy:write    # Required only if also uploading lookup files

response = falcon.perform_action(action_name="contain", ids=host_ids)

if response["status_code"] == 207:
    errors = response.get("body", {}).get("errors", [])
    rate_limited = [e for e in errors if e.get("code") == 429]
    if rate_limited:
        return Response(body={"error": "Rate limited", "failed_ids": [e.get("id") for e in rate_limited]}, code=429)

def test_get_alerts_success():
    mock_falcon = Mock()
    mock_falcon.query_alerts_v2.return_value = {
        "status_code": 200,
        "body": {"resources": ["alert-001", "alert-002"]}
    }
    mock_falcon.get_alerts_v2.return_value = {
        "status_code": 200,
        "body": {"resources": [{"id": "alert-001", "severity": 80}]}
    }

    with patch('falconpy.Alerts', return_value=mock_falcon):
        from main import get_alerts
        request = Mock(spec=Request)
        request.params = {"limit": "10"}
        response = get_alerts(request, None, Mock())
        assert response.code == 200
        assert len(response.body["alerts"]) == 1

export FALCON_CLIENT_ID="your-client-id"
export FALCON_CLIENT_SECRET="your-client-secret"
cd functions/my-function && python3 main.py
curl -X GET http://localhost:8081/api/alerts?limit=10

# manifest.yml
auth:
    scopes:
        - devices:read
        - falconx-indicators:read
    permissions: {}
    roles: []