### Python: Login and Get All Devices

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/_sources/docs/device-manager/get-devices-no-filter.md

Complete Python example that logs into FortiAnalyzer and retrieves all devices. Includes functions for loading config, logging in, logging out, and fetching devices from a specific ADOM. Uses requests with SSL verification disabled and handles API errors.

```python
import json
import requests
import urllib3

# Disable SSL warnings for self-signed certificates
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

def load_config():
    """Load FortiAnalyzer configuration from .faz-env.json"""
    with open('.faz-env.json', 'r') as f:
        return json.load(f)

def login(config):
    """Establish session with FortiAnalyzer"""
    url = f"https://{config['faz_host']}:{config['faz_port']}/jsonrpc"

    payload = {
        "method": "exec",
        "params": [{
            "url": "/sys/login/user",
            "data": {
                "user": config['username'],
                "passwd": config['password']
            }
        }],
        "session": None,
        "id": 1
    }

    try:
        response = requests.post(url, json=payload, verify=False, timeout=10)
        response.raise_for_status()
        result = response.json()

        if result['result'][0]['status']['code'] == 0:
            return result.get('session')
        else:
            raise Exception(f"Login failed: {result['result'][0]['status']['message']}")
    except requests.exceptions.RequestException as e:
        raise Exception(f"Connection error: {str(e)}")

def logout(config, session_id):
    """Terminate session with FortiAnalyzer"""
    url = f"https://{config['faz_host']}:{config['faz_port']}/jsonrpc"

    payload = {
        "method": "exec",
        "params": [{"url": "/sys/logout"}],
        "session": session_id,
        "id": 999
    }

    requests.post(url, json=payload, verify=False)

def get_devices_in_adom(session_id, config, adom="root"):
    """
    Get all devices in a specific ADOM

    Args:
        session_id: Active session ID
        config: Configuration dictionary
        adom: ADOM name (default: "root")

    Returns:
        list: List of device dictionaries
    """
    url = f"https://{config['faz_host']}:{config['faz_port']}/jsonrpc"

    payload = {
        "method": "get",
        "params": [{
            "url": f"/dvmdb/adom/{adom}/device"
        }],
        "session": session_id,
        "id": 2
    }

    try:
        response = requests.post(url, json=payload, verify=False, timeout=30)
        response.raise_for_status()
        result = response.json()

        if result['result'][0]['status']['code'] == 0:
            return result['result'][0]['data']
        else:
            raise Exception(f"API error: {result['result'][0]['status']['message']}")
    except requests.exceptions.RequestException as e:
        raise Exception(f"Request error: {str(e)}")

def get_all_devices(session_id, config):
    """
    Get all devices across all ADOMs

    Args:
        session_id: Active session ID
        config: Configuration dictionary

    Returns:
        list: List of all device dictionaries
    """
    url = f"https://{config['faz_host']}:{config['faz_port']}/jsonrpc"

    payload = {
        "method": "get",
        "params": [{
            "url": "/dvmdb/device"
        }],
        "session": session_id,
        "id": 3
    }

    try:
        response = requests.post(url, json=payload, verify=False, timeout=30)
        response.raise_for_status()
        result = response.json()

        if result['result'][0]['status']['code'] == 0:
            return result['result'][0]['data']
        else:
            raise Exception(f"API error: {result['result'][0]['status']['message']}")
    except requests.exceptions.RequestException as e:
        raise Exception(f"Request error: {str(e)}")
```

--------------------------------

### Example: Get all Debian endpoints

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/_sources/docs/fabric-viewasset-identity-center/get-ueba-endpoints-by-os.md

Calls get_ueba_endpoints_by_os to retrieve Debian endpoints within a specified time range, then prints hostname, IP, OS version, risk score, and last seen for each endpoint.

```python
# Example: Get all Debian endpoints
endpoints = get_ueba_endpoints_by_os(
    session_id="your_session_id",
    adom="root",
    os_name="Debian",
    time_range={
        "start": "2023-10-06 13:09:00",
        "end": "2023-12-05 13:09:00"
    }
)

# Display results
print(f"Found {len(endpoints)} Debian endpoints:\n")
for ep in endpoints:
    print(f"  {ep['hostname']} ({ep['ip']})")
    print(f"    OS Version: {ep['osversion']}")
    print(f"    Risk Score: {ep['risk_score']}")
    print(f"    Last Seen: {ep['last_seen']}")
    print()
```

--------------------------------

### Complete Python Example: Get Automation Connectors

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/_sources/docs/incidents-eventsautomation-connectors/get-fabric-conector.md

Defines get_automation_connectors() to fetch all webhook connectors from FortiAnalyzer. It sends a POST request to the /jsonrpc endpoint with a session ID and ADOM, and returns the connector data. The example usage prints each connector's name, URL, protocol, port, and status.

```python
import requests
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

def get_automation_connectors(session_id, adom="root"):
    """
    Get all automation connectors (webhooks)

    Args:
        session_id: Active session ID
        adom: ADOM name (default: "root")

    Returns:
        list: Automation connector configurations
    """
    url = "https://faz.example.com/jsonrpc"

    payload = {
        "method": "get",
        "params": [{
            "url": f"/config/adom/{adom}/system/webhook"
        }],
        "session": session_id,
        "id": 1
    }

    response = requests.post(url, json=payload, verify=False)
    result = response.json()

    if result['result'][0]['status']['code'] == 0:
        return result['result'][0]['data']
    else:
        raise Exception(f"API error: {result['result'][0]['status']['message']}")

# Example usage
connectors = get_automation_connectors(session_id="your_session_id")

print(f"Total Automation Connectors: {len(connectors)}\n")

for conn in connectors:
    status_icon = "✓" if conn['status'] == 1 else "✗"
    print(f"{status_icon} {conn['name']} ({conn['title']})")
    print(f"   URL: {conn['uri']}")
    print(f"   Protocol: {conn['protocol']}:{conn['port']}")
    print(f"   Status: {'Enabled' if conn['status'] == 1 else 'Disabled'}")
    print()
```

--------------------------------

### Complete Python Example: Get Top Sources

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/_sources/docs/fortiviewtop-sources/create-task.md

This Python script submits a task to the FortiView Top Sources API, polls for completion, and prints the top sources. It uses the requests library with SSL verification disabled and includes a time range and sort field. The example shows how to call the get_top_sources function with a session ID, ADOM, and time range, then displays the results with session, byte, and bandwidth counts.

```python
import json
import requests
import urllib3
import time

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

def get_top_sources(session_id, adom, time_range, limit=10, sort_field="sessions"):
    """
    Get top traffic sources for a time period

    Args:
        session_id: Active session ID
        adom: ADOM name
        time_range: Time range dict with 'start' and 'end'
        limit: Number of top sources to return (default: 10)
        sort_field: Field to sort by: sessions, bytes, bandwidth

    Returns:
        list: Top sources data
    """
    url = "https://faz.example.com/jsonrpc"

    # Step 1: Submit task
    payload = {
        "method": "add",
        "params": [{
            "url": f"/fortiview/adom/{adom}/top-sources/run",
            "apiver": 3,
            "case-sensitive": False,
            "device": [{"devid": "All_Devices"}],
            "filter": "",
            "limit": limit,
            "sort-by": [{
                "field": sort_field,
                "order": "desc"
            }],
            "time-range": time_range
        }],
        "session": session_id,
        "id": 1
    }

    response = requests.post(url, json=payload, verify=False)
    result = response.json()

    tid = result['result'][0]['data']['tid']
    print(f"✓ Task submitted. TID: {tid}")

    # Step 2: Poll for completion
    while True:
        poll_payload = {
            "method": "get",
            "params": [{
                "url": f"/fortiview/adom/{adom}/top-sources/{tid}"
            }],
            "session": session_id,
            "id": 2
        }

        response = requests.post(url, json=poll_payload, verify=False)
        data = response.json()['result'][0]['data']

        if data['status'] == 'done' and data['percentage'] == 100:
            print(f"✓ Found {data['total']} top sources")
            return data.get('sources', [])

        time.sleep(2)

# Example: Get top 10 sources by session count
sources = get_top_sources(
    session_id="your_session_id",
    adom="root",
    time_range={
        "start": "2025-11-09 00:00:00",
        "end": "2025-11-09 23:59:59"
    },
    limit=10,
    sort_field="sessions"
)

# Display results
print("\nTop Traffic Sources:")
for i, src in enumerate(sources, 1):
    print(f"{i}. {src['srcip']} ({src.get('hostname', 'Unknown')})")
    print(f"   Sessions: {src['sessions']:,}")
    print(f"   Bytes: {src['bytes']:,}")
    print(f"   Bandwidth: {src['bandwidth']:,} bps")
    print()
```

--------------------------------

### Get All Applications to See Policy Names

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/_sources/docs/fortiviewtop-applications/topapplications-w-policyname.md

Python example that queries all top applications without a policy filter to extract and print unique policy names. Useful for finding exact policy names before filtering.

```python
# Get all applications to see policy names
all_apps = get_top_applications(
    session_id=session,
    adom="root",
    time_range={"last-n-hours": 24},
    limit=1000
)

# Extract unique policy names
policy_names = set(app['policyname'] for app in all_apps)
print("Available policy names:")
for pname in sorted(policy_names):
    print(f"  - {pname}")
```

--------------------------------

### Python: Get report schedules

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/docs/reports/get-report-schedules.html

Complete Python example using requests to retrieve report schedules. It disables SSL warnings, sends a POST request, and raises an exception on API error. The example prints schedule details.

```python
import requests
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

def get_report_schedules(session_id, adom="root"):
    """Get configured report schedules"""
    url = "https://faz.example.com/jsonrpc"
    
    payload = {
        "method": "get",
        "params": [{
            "url": f"/config/adom/{adom}/sql-report/schedule"
        }],
        "session": session_id,
        "id": 1
    }
    
    response = requests.post(url, json=payload, verify=False)
    result = response.json()
    
    if result['result'][0]['status']['code'] == 0:
        return result['result'][0]['data']
    else:
        raise Exception(f"API error: {result['result'][0]['status']['message']}")

# Example
schedules = get_report_schedules(session_id="your_session_id")
for schedule in schedules:
    print(f"Schedule: {schedule['schedule-name']}")
    print(f"  Type: {schedule['schedule-type']}")
    print(f"  Time: {schedule.get('time', 'N/A')}")
```

--------------------------------

### Request Example: POST /jsonrpc with /logview/adom/{adom}/logsearch/{tid}

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/docs/logview/fetch-log-search-result-by-task-id.html

Example request payload for fetching log search results. Uses method 'get', URL '/logview/adom/root/logsearch/12345', and includes limit and offset parameters. The session ID is a placeholder.

```json
{
    "method": "get",
    "params": [{
        "url": "/logview/adom/root/logsearch/12345",
        "data": {
            "limit": 100,
            "offset": 0
        }
    }],
    "session": "{{session_id}}",
    "id": 1
}
```

--------------------------------

### Complete Python Example

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/docs/logview/create-search-task-for-attack---botnet.html

Python example demonstrating the two-step asynchronous pattern for searching botnet logs using the FortiAnalyzer JSON-RPC API.

```APIDOC
## POST /jsonrpc

### Description
Submits a botnet-related attack log search request to the FortiAnalyzer LogView API. This is the first step of a two-step asynchronous search pattern.

### Method
POST

### Endpoint
/jsonrpc

### Parameters
#### Request Body
- **method** (string) - Required - The JSON-RPC method to invoke. Must be `add` for this step.
- **params** (array) - Required - An array containing a single parameter object.
- **params[0].url** (string) - Required - The API path for the search request. Format: `/logview/adom/{adom}/logsearch`.
- **params[0].data.logtype** (string) - Required - The type of logs to search. Must be `attack`.
- **params[0].data.filter** (string) - Required - The filter expression to apply. For botnet detection, use `attack contains "botnet"`.
- **params[0].data.time-range** (object) - Required - The time range for the search.
- **params[0].data.time-range.start** (string) - Required - The start time in `YYYY-MM-DD HH:MM:SS` format.
- **params[0].data.time-range.end** (string) - Required - The end time in `YYYY-MM-DD HH:MM:SS` format.
- **session** (string) - Required - The session ID for authentication.
- **id** (integer) - Required - A unique request ID.

### Request Example
```json
{
    "method": "add",
    "params": [{
        "url": "/logview/adom/root/logsearch",
        "data": {
            "logtype": "attack",
            "filter": "attack contains \"botnet\"",
            "time-range": {
                "start": "2025-11-09 00:00:00",
                "end": "2025-11-09 23:59:59"
            }
        }
    }],
    "session": "{{session_id}}",
    "id": 1
}
```

### Response
#### Success Response (200)
- **result[0].data.tid** (integer) - The task ID for the search, used in the next step.
- **result[0].status.code** (integer) - The status code. `0` indicates success.
- **result[0].status.message** (string) - The status message.

#### Response Example
```json
{
    "result": [{
        "data": {
            "tid": 12351
        },
        "status": {
            "code": 0,
            "message": "OK"
        }
    }]
}
```
```

```APIDOC
## POST /jsonrpc

### Description
Fetches the results of a previously submitted botnet log search. This is the second step of the two-step asynchronous search pattern.

### Method
POST

### Endpoint
/jsonrpc

### Parameters
#### Request Body
- **method** (string) - Required - The JSON-RPC method to invoke. Must be `get` for this step.
- **params** (array) - Required - An array containing a single parameter object.
- **params[0].url** (string) - Required - The API path for fetching results. Format: `/logview/adom/{adom}/logsearch/{tid}`.
- **params[0].data.limit** (integer) - Optional - The maximum number of log entries to return.
- **params[0].data.offset** (integer) - Optional - The offset for pagination.
- **session** (string) - Required - The session ID for authentication.
- **id** (integer) - Required - A unique request ID.

### Request Example
```json
{
    "method": "get",
    "params": [{
        "url": "/logview/adom/root/logsearch/12351",
        "data": {
            "limit": 100,
            "offset": 0
        }
    }],
    "session": "{{session_id}}",
    "id": 2
}
```

### Response
#### Success Response (200)
- **result[0].data.tid** (integer) - The task ID of the search.
- **result[0].data.status** (string) - The status of the search. `done` indicates completion.
- **result[0].data.percentage** (integer) - The completion percentage. `100` indicates complete.
- **result[0].data.total_lines** (integer) - The total number of log entries found.
- **result[0].data.logs** (array) - An array of log entries.
- **result[0].data.logs[].date** (string) - The date of the log entry.
- **result[0].data.logs[].time** (string) - The time of the log entry.
- **result[0].data.logs[].devname** (string) - The device name.
- **result[0].data.logs[].srcip** (string) - The source IP address.
- **result[0].data.logs[].dstip** (string) - The destination IP address.
- **result[0].data.logs[].attack** (string) - The attack name.
- **result[0].data.logs[].action** (string) - The action taken.
- **result[0].data.logs[].severity** (string) - The severity level.
- **result[0].status.code** (integer) - The status code. `0` indicates success.
- **result[0].status.message** (string) - The status message.

#### Response Example
```json
{
    "result": [{
        "data": {
            "tid": 12351,
            "status": "done",
            "percentage": 100,
            "total_lines": 12,
            "logs": [
                {
                    "date": "2025-11-09",
                    "time": "17:05:23",
                    "devname": "FGT-01",
                    "srcip": "10.0.1.150",
                    "dstip": "198.51.100.45",
                    "attack": "Botnet.CnC.Connection",
                    "action": "dropped",
                    "severity": "high"
                }
            ]
        },
        "status": {
            "code": 0,
            "message": "OK"
        }
    }]
}
```
```

--------------------------------

### cURL example configuration variables

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/_sources/docs/device-manager/get-adom-with-no-fields.md

Bash script header for a cURL example that configures FortiAnalyzer connection parameters. Placeholder values for host, port, username, and password must be replaced with actual credentials.

```bash
#!/bin/bash

# Configuration
FAZ_HOST="faz.example.com"
FAZ_PORT="443"
USERNAME="admin"
PASSWORD="your_password_here"
```

--------------------------------

### Response: Get report schedules

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/docs/reports/get-report-schedules.html

Example response showing a list of report schedules. The status code 0 indicates success.

```json
{
    "result": [{
        "data": [
            {
                "id": 1,
                "layout-id": 123456,
                "schedule-name": "Daily_Traffic_Report",
                "schedule-type": "daily",
                "time": "08:00"
            }
        ],
        "status": {
            "code": 0,
            "message": "OK"
        }
    }]
}
```

--------------------------------

### Request Example for /sys/status

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/_sources/docs/pilot/system-status.md

JSON-RPC request to retrieve system status. Uses the 'get' method with the '/sys/status' URL and a session ID placeholder.

```json
{
    "method": "get",
    "params": [{
        "url": "/sys/status"
    }],
    "session": "{{session_id}}",
    "id": 1
}
```

--------------------------------

### Response Example for Fetching Top Sources Results

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/_sources/docs/fortiviewtop-sources/fetch-result-by-task.md

Example response showing a completed task with status 'done', 100% completion, and a list of top sources with fields like srcip, hostname, sessions, bytes, bandwidth, and country.

```json
{
    "result": [{
        "data": {
            "tid": 12450,
            "status": "done",
            "percentage": 100,
            "total": 10,
            "sources": [
                {
                    "srcip": "10.0.1.150",
                    "hostname": "LAPTOP-USER01",
                    "sessions": 4523,
                    "bytes": 2147483648,
                    "bandwidth": 45678901,
                    "country": "Reserved"
                },
                {
                    "srcip": "10.0.1.75",
                    "hostname": "DESKTOP-IT01",
                    "sessions": 3891,
                    "bytes": 1879048192,
                    "bandwidth": 39012345,
                    "country": "Reserved"
                },
                {
                    "srcip": "10.0.1.45",
                    "hostname": "SERVER-APP01",
                    "sessions": 3204,
                    "bytes": 1610612736,
                    "bandwidth": 33456789,
                    "country": "Reserved"
                }
            ]
        },
        "status": {
            "code": 0,
            "message": "OK"
        }
    }]
}
```

--------------------------------

### Conditional disable of non-critical event handlers

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/_sources/docs/incidents-eventsevent-handlers-setup/disable-eventhandler.md

Starts an example of disabling only non-critical event handlers by importing get_event_handlers. The snippet is incomplete as displayed.

```python
# Disable only non-critical event handlers
from get_eventhandler import get_event_handlers
```

--------------------------------

### Complete Python Example - search_webfilter_logs

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/docs/logview/create-search-task-for-webfilter-logs.html

Complete Python example that submits a webfilter log search, polls for completion, and prints the first 10 logs. Uses requests and urllib3, disables SSL warnings, and raises an exception on search failure.

```python
import requests
import urllib3
import time

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

def search_webfilter_logs(session_id, adom, filter_expr, time_range):
    """
    Search webfilter logs

    Args:
        session_id: Active session ID
        adom: ADOM name
        filter_expr: Filter expression (e.g., "action=blocked")
        time_range: Time range dict

    Returns:
        list: Matching webfilter log entries
    """
    url = "https://faz.example.com/jsonrpc"

    # Step 1: Submit search
    payload = {
        "method": "add",
        "params": [{
            "url": f"/logview/adom/{adom}/logsearch",
            "data": {
                "logtype": "webfilter",
                "filter": filter_expr,
                "time-range": time_range
            }
        }],
        "session": session_id,
        "id": 1
    }

    response = requests.post(url, json=payload, verify=False)
    result = response.json()

    if result['result'][0]['status']['code'] != 0:
        raise Exception(f"Search failed: {result['result'][0]['status']['message']}")

    tid = result['result'][0]['data']['tid']
    print(f"✓ Webfilter search submitted. TID: {tid}")

    # Step 2: Poll and fetch
    while True:
        payload = {
            "method": "get",
            "params": [{
                "url": f"/logview/adom/{adom}/logsearch/{tid}"
            }],
            "session": session_id,
            "id": 2
        }

        response = requests.post(url, json=payload, verify=False)
        data = response.json()['result'][0]['data']

        if data['status'] == 'done' and data['percentage'] == 100:
            print(f"✓ Found {data['total_lines']} webfilter logs")
            return data.get('logs', [])

        print(f"  Status: {data['percentage']}%")
        time.sleep(2)

# Example: Find all blocked websites in last 24 hours
logs = search_webfilter_logs(
    session_id="your_session_id",
    adom="root",
    filter_expr="action=blocked",
    time_range={"last-n-hours": 24}
)

# Display summary
for log in logs[:10]:
    print(f"{log['time']} | {log.get('user', 'unknown')} -> "
          f"{log['hostname']} | Category: {log.get('cat desc', 'N/A')}")

```

--------------------------------

### OS Migration Planning

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/_sources/docs/fabric-viewasset-identity-center/get-ueba-endpoints-by-os.md

Gets endpoints running legacy OS versions (Windows 7, Windows Server 2008, Ubuntu 16.04) and prints a migration report with hostname, OS name, and version.

```python
# Get all endpoints running legacy OS versions
legacy_os = ["Windows 7", "Windows Server 2008", "Ubuntu 16.04"]
endpoints_to_migrate = []

for os_name in legacy_os:
    endpoints = get_ueba_endpoints_by_os(
        session_id=session,
        adom="root",
        os_name=os_name
    )
    endpoints_to_migrate.extend(endpoints)

print(f"Total endpoints requiring migration: {len(endpoints_to_migrate)}")

# Generate migration report
for ep in endpoints_to_migrate:
    print(f"{ep['hostname']}: {ep['osname']} {ep['osversion']} -> Upgrade Required")
```

--------------------------------

### Response: Get report templates

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/docs/reports/get-report-templates.html

Example response containing a list of report templates with name, description, and type. Status code 0 indicates success.

```json
{
    "result": [{
        "data": [
            {
                "name": "Daily_Traffic_Report",
                "description": "Daily traffic summary",
                "type": "sql-report"
            }
        ],
        "status": {
            "code": 0,
            "message": "OK"
        }
    }]
}
```

--------------------------------

### GET /fortiview/adom/root/top-applications/run/12458 - Request

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/_sources/docs/fortiviewtop-applications/topapplications-w-policyname.md

Second step to fetch the results of the previously started top-applications query. Uses the 'get' method with the task ID from the first response. The response contains the application list with details like app name, category, sessions, bytes, bandwidth, and policy name.

```json
{
    "method": "get",
    "params": [{
        "url": "/fortiview/adom/root/top-applications/run/12458"
    }],
    "session": "{{session_id}}",
    "id": 2
}
```

```json
{
    "result": [{
        "data": {
            "tid": 12458,
            "status": "done",
            "percentage": 100,
            "total": 12,
            "applications": [
                {
                    "app": "Office365.Sharepoint",
                    "appcat": "Cloud.IT",
                    "sessions": 3245,
                    "bytes": 1073741824,
                    "bandwidth": 22345678,
                    "policyid": 125,
                    "policyname": "Intra-2-SDWAN_BBI"
                },
                {
                    "app": "Microsoft.Teams",
                    "appcat": "Collaboration",
                    "sessions": 2891,
                    "bytes": 805306368,
                    "bandwidth": 16789012,
                    "policyid": 125,
                    "policyname": "Intra-2-SDWAN_BBI"
                }
            ]
        },
        "status": {
            "code": 0,
            "message": "OK"
        }
    }]
}
```

--------------------------------

### Complete Python Example: get_managed_devices

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/_sources/docs/system-settings/get-info-from-managed-devices.md

Python function using requests to call the FortiAnalyzer JSON-RPC API. Disables SSL warnings, posts to /jsonrpc, and raises an exception on API error. Example usage prints device name, serial, IP, and version.

```python
import requests
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

def get_managed_devices(session_id, adom="root"):
    """Get managed device information"""
    url = "https://faz.example.com/jsonrpc"

    payload = {
        "method": "get",
        "params": [{"url": f"/dvmdb/adom/{adom}/device"}],
        "session": session_id,
        "id": 1
    }

    response = requests.post(url, json=payload, verify=False)
    result = response.json()

    if result['result'][0]['status']['code'] == 0:
        return result['result'][0]['data']
    else:
        raise Exception(f"API error: {result['result'][0]['status']['message']}")

# Example
devices = get_managed_devices(session_id="your_session_id")
for device in devices:
    print(f"Device: {device['name']} ({device['sn']})")
    print(f"  IP: {device['ip']}")
    print(f"  Version: {device['os_ver']}")
```

--------------------------------

### Python Example for System Status

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/_sources/docs/pilot/system-status.md

Complete Python example that loads configuration, logs in, and retrieves system status. It disables SSL warnings for self-signed certificates and raises exceptions on login or connection errors.

```python
import json
import requests
import urllib3

# Disable SSL warnings for self-signed certificates
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

def load_config():
    """Load FortiAnalyzer configuration from .faz-env.json"""
    with open('.faz-env.json', 'r') as f:
        return json.load(f)

def login(config):
    """Establish session with FortiAnalyzer"""
    url = f"https://{config['faz_host']}:{config['faz_port']}/jsonrpc"
    
    payload = {
        "method": "exec",
        "params": [{
            "url": "/sys/login/user",
            "data": {
                "user": config['username'],
                "passwd": config['password']
            }
        }],
        "session": None,
        "id": 1
    }
    
    try:
        response = requests.post(url, json=payload, verify=False, timeout=10)
        response.raise_for_status()
        result = response.json()
        
        if result['result'][0]['status']['code'] == 0:
            return result.get('session')
        else:
            raise Exception(f"Login failed: {result['result'][0]['status']['message']}")
    except requests.exceptions.RequestException as e:
        raise Exception(f"Connection error: {str(e)}")
```

--------------------------------

### Get Event Handlers - Python Example

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/_sources/docs/incidents-eventsevent-handlers-setup/get-eventhandler.md

Complete Python example using the requests library to call the /jsonrpc endpoint and retrieve event handlers. It disables SSL warnings, sends a POST request with the payload, and checks the status code. On success, it returns the data list; otherwise, it raises an exception with the API error message. The example then prints the total count and details for each handler.

```python
import requests
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

def get_event_handlers(session_id, adom="root"):
    """Get all event handlers"""
    url = "https://faz.example.com/jsonrpc"

    payload = {
        "method": "get",
        "params": [{
            "url": f"/eventmgmt/adom/{adom}/conf-eventhandler"
        }],
        "session": session_id,
        "id": 1
    }

    response = requests.post(url, json=payload, verify=False)
    result = response.json()

    if result['result'][0]['status']['code'] == 0:
        return result['result'][0]['data']
    else:
        raise Exception(f"API error: {result['result'][0]['status']['message']}")

# Example
handlers = get_event_handlers(session_id="your_session_id")

print(f"Total Event Handlers: {len(handlers)}\n")
for handler in handlers:
    status_icon = "✓" if handler.get('status') == 'enabled' else "✗"
    print(f"{status_icon} {handler['name']}")
    print(f"   Trigger: {handler.get('trigger')}")
    print(f"   Action: {handler.get('action')}")
    print()
```

--------------------------------

### Response Example for /sys/status

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/_sources/docs/pilot/system-status.md

Example response from the System Status endpoint, showing system information, disk usage, memory usage, and status.

```json
{
    "result": [{
        "data": {
            "Admin Domain Configuration": "Enabled",
            "BIOS version": "04.06.05",
            "Branch Point": "2086",
            "Build": "2086",
            "Current Time": "Sun Nov  9 14:30:00 2024",
            "Disk Usage": [
                {
                    "name": "/",
                    "total": 102932480,
                    "used": 15439872
                },
                {
                    "name": "/data",
                    "total": 1948374016,
                    "used": 245760000
                }
            ],
            "Hostname": "FAZ-PROD-01",
            "License Status": "Valid",
            "Memory Usage": {
                "total": 16384,
                "used": 8192
            },
            "Model": "FortiAnalyzer-VM64",
            "Serial Number": "FAZVM1234567890",
            "Version": "v7.6.4-build2086 250929 (GA)"
        },
        "status": {
            "code": 0,
            "message": "OK"
        },
        "url": "/sys/status"
    }],
    "session": "kLSFZ7qT9xZc1vB3rtKlXg==",
    "id": 1
}
```

--------------------------------

### Session-Based Authentication: Login, Use, and Logout

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/getting-started/index.html

Logs in to get a session ID, uses it in subsequent requests, and logs out when done. The example uses verify=False and a placeholder password.

```python
# Login to get session ID
login_response = requests.post(url, json={
    "method": "exec",
    "params": [{
        "url": "/sys/login/user",
        "data": {"user": "admin", "passwd": "password"}
    }],
    "session": None,
    "id": 1
}, verify=False)

session_id = login_response.json()['session']

# Use session ID in subsequent requests
# ... API calls ...

# Always logout when done
requests.post(url, json={
    "method": "exec",
    "params": [{"url": "/sys/logout"}],
    "session": session_id,
    "id": 999
}, verify=False)

```

--------------------------------

### Response: Get Fabric Connector Event Handler

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/docs/incidents-eventsevent-handlers-setup/get-fabric-connector-eventhandler.html

Example response showing two enabled connectors (Splunk_HEC, ServiceNow_Incidents) with status code 0 and message OK.

```json
{
    "result": [{
        "data": [
            {
                "name": "Splunk_HEC",
                "status": "enabled"
            },
            {
                "name": "ServiceNow_Incidents",
                "status": "enabled"
            }
        ],
        "status": {
            "code": 0,
            "message": "OK"
        },
        "url": "/eventmgmt/adom/root/config/trigger/1/fabric-connector"
    }],
    "session": "{{session_id}}",
    "id": 1
}
```

--------------------------------

### Python: get_ueba_endpoints_by_os function and usage

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/docs/fabric-viewasset-identity-center/get-ueba-endpoints-by-os.html

Complete Python example that defines a function to retrieve UEBA endpoints filtered by OS name, then demonstrates usage with a time range and prints results. Uses requests and urllib3, disables SSL warnings, and raises an exception on API error.

```python
import json
import requests
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

def get_ueba_endpoints_by_os(session_id, adom, os_name, time_range=None, limit=1000):
    """
    Get UEBA endpoints filtered by operating system

    Args:
        session_id: Active session ID
        adom: ADOM name
        os_name: Operating system name to filter
        time_range: Optional time range dict
        limit: Maximum results (default: 1000)

    Returns:
        list: Endpoint data
    """
    url = "https://faz.example.com/jsonrpc"

    params_data = {
        "url": f"/ueba/adom/{adom}/endpoints/",
        "filter": f"osname={os_name}",
        "limit": limit,
        "offset": 0,
        "detail-level": "standard",
        "sort-by": [{
            "field": "epid",
            "order": "asc"
        }]
    }

    if time_range:
        params_data["time-range"] = time_range

    payload = {
        "method": "get",
        "params": [params_data],
        "session": session_id,
        "id": 1
    }

    response = requests.post(url, json=payload, verify=False)
    result = response.json()

    if result['result'][0]['status']['code'] == 0:
        return result['result'][0].get('data', [])
    else:
        raise Exception(f"API error: {result['result'][0]['status']['message']}")

# Example: Get all Debian endpoints
endpoints = get_ueba_endpoints_by_os(
    session_id="your_session_id",
    adom="root",
    os_name="Debian",
    time_range={
        "start": "2023-10-06 13:09:00",
        "end": "2023-12-05 13:09:00"
    }
)

# Display results
print(f"Found {len(endpoints)} Debian endpoints:\n")
for ep in endpoints:
    print(f"  {ep['hostname']} ({ep['ip']})")
    print(f"    OS Version: {ep['osversion']}")
    print(f"    Risk Score: {ep['risk_score']}")
    print(f"    Last Seen: {ep['last_seen']}")
    print()
```

--------------------------------

### Complete Python Example: Get Top Applications

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/_sources/docs/fortiviewtop-applications/topapplications.md

Python script that submits a task to the FortiView Top Applications API, polls for completion, and prints the top applications by bandwidth. It uses the /fortiview/adom/{adom}/top-applications/run endpoint with a time range and optional filter, then displays the top 10 results with session, bytes, bandwidth, and policy information.

```python
import json
import requests
import urllib3
import time

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

def get_top_applications(session_id, adom, time_range, limit=100, filter_expr=""):
    """
    Get top applications for a time period

    Args:
        session_id: Active session ID
        adom: ADOM name
        time_range: Time range dict with 'start' and 'end'
        limit: Number of top applications to return (default: 100)
        filter_expr: Optional filter expression (e.g., "policyid=46")

    Returns:
        list: Top applications data
    """
    url = "https://faz.example.com/jsonrpc"

    # Step 1: Submit task
    payload = {
        "method": "add",
        "params": [{
            "url": f"/fortiview/adom/{adom}/top-applications/run",
            "apiver": 3,
            "case-sensitive": False,
            "device": [{"devid": "All_Devices"}],
            "filter": filter_expr,
            "limit": limit,
            "sort-by": [{
                "field": "bytes",
                "order": "desc"
            }],
            "time-range": time_range
        }],
        "session": session_id,
        "id": 1
    }

    response = requests.post(url, json=payload, verify=False)
    result = response.json()

    tid = result['result'][0]['data']['tid']
    print(f"✓ Task submitted. TID: {tid}")

    # Step 2: Poll for completion
    while True:
        poll_payload = {
            "method": "get",
            "params": [{
                "url": f"/fortiview/adom/{adom}/top-applications/run/{tid}"
            }],
            "session": session_id,
            "id": 2
        }

        response = requests.post(url, json=poll_payload, verify=False)
        data = response.json()['result'][0]['data']

        if data['status'] == 'done' and data['percentage'] == 100:
            print(f"✓ Found {data['total']} top applications")
            return data.get('applications', [])

        time.sleep(2)

# Example: Get top applications by bandwidth
apps = get_top_applications(
    session_id="your_session_id",
    adom="root",
    time_range={
        "start": "2025-11-09 00:00:00",
        "end": "2025-11-09 23:59:59"
    },
    limit=100,
    filter_expr="policyid=46"
)

# Display results
print("\nTop Applications by Bandwidth:")
for i, app in enumerate(apps[:10], 1):
    print(f"{i}. {app['app']} ({app['appcat']})")
    print(f"   Sessions: {app['sessions']:,}")
    print(f"   Bytes: {app['bytes']:,} ({app['bytes']/1024/1024/1024:.2f} GB)")
    print(f"   Bandwidth: {app['bandwidth']/1000000:.2f} Mbps")
    print(f"   Policy: {app['policyname']} (ID: {app['policyid']})")
    print()
```

--------------------------------

### API Key Authentication: Use Bearer Token in Authorization Header

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/getting-started/index.html

Uses an API key in the Authorization header for requests. The example sends a get request to /sys/status with verify=False.

```python
# Use API key in Authorization header
headers = {
    "Authorization": "Bearer your_api_key_here",
    "Content-Type": "application/json"
}

response = requests.post(url, json={
    "method": "get",
    "params": [{"url": "/sys/status"}],
    "session": None,
    "id": 1
}, headers=headers, verify=False)

```

--------------------------------

### Complete Python Example to Get Certificates

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/_sources/docs/system-settings/get-cert.md

Python function get_certificates(session_id) sends a POST request to the /jsonrpc endpoint to retrieve certificates. It disables SSL warnings, uses verify=False, and raises an exception if the API returns a non-zero status code. The example prints each certificate's name, subject, and issuer.

```python
import requests
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

def get_certificates(session_id):
    """Get installed certificates"""
    url = "https://faz.example.com/jsonrpc"

    payload = {
        "method": "get",
        "params": [{"url": "/cli/global/system/certificate/local"}],
        "session": session_id,
        "id": 1
    }

    response = requests.post(url, json=payload, verify=False)
    result = response.json()

    if result['result'][0]['status']['code'] == 0:
        return result['result'][0]['data']
    else:
        raise Exception(f"API error: {result['result'][0]['status']['message']}")

# Example
certs = get_certificates(session_id="your_session_id")
for cert in certs:
    print(f"Certificate: {cert['name']}")
    print(f"  Subject: {cert['subject']}")
    print(f"  Issuer: {cert['issuer']}")
```

--------------------------------

### Download Report Request Example

Source: https://how-to-fortianalyzer-api.readthedocs.io/en/latest/docs/reports/download-report.html

JSON-RPC request to download a report. Uses the 'get' method with the URL '/report/adom/root/reports/12345' and a session ID placeholder. The response contains the report data in base64 encoding.

```json
{
    "method": "get",
    "params": [{
        "url": "/report/adom/root/reports/12345"
    }],
    "session": "{{session_id}}",
    "id": 1
}
```