Metadata-Version: 2.4
Name: aiq-platform-api
Version: 1.0.77
Summary: Utility functions for AttackIQ Platform API usage
Author: Rajesh Sharma
Author-email: Rajesh Sharma <rajesh.sharma@attackiq.com>
License-Expression: MIT
License-File: LICENSE
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Requires-Dist: python-dotenv>=1.0.1,<2.0.0
Requires-Dist: httpx>=0.27,<1.0
Requires-Dist: tenacity>=8.2.3
Requires-Dist: ipython>=7.34.0,<10.0.0
Requires-Dist: pyzipper>=0.4.0,<0.5.0
Requires-Python: >=3.11, <3.15
Description-Content-Type: text/markdown

# AttackIQ Platform API

> ⚠️ **Beta** - Under active development. APIs subject to change. Feedback: rajesh.sharma@attackiq.com | Access: Request invite to AttackIQ GitHub.

Tools for interacting with the AttackIQ Platform API:
- **Python SDK** (`aiq-platform-api`) - Async library for Python applications
- **CLI** (`aiq`) - Command-line interface

---

## Python SDK

Install from PyPI:

```sh
pip install aiq-platform-api
```

### Usage

```python
import asyncio
from aiq_platform_api import AttackIQClient, Scenarios, Assets

async def main():
    async with AttackIQClient(
        "https://your-platform.attackiq.com",
        "your-api-token"
    ) as client:
        # Search scenarios
        result = await Scenarios.search_scenarios(client, query="powershell", limit=10)
        print(f"Found {result['count']} scenarios")

        # List assets
        async for asset in Assets.get_assets(client, limit=5):
            print(asset["hostname"])

asyncio.run(main())
```

### Automatic HTTP 429 Handling

When an SDK-managed request receives HTTP 429, the SDK retries with bounded exponential backoff and full jitter:

- Each request has at most five total attempts and a 60-second retry budget, including requests and waits.
  No new attempt starts after the budget expires. Each pagination page receives its own budget.
- Each asynchronous wait is capped at 15 seconds. A valid `Retry-After` (seconds or an HTTP date) sets the minimum
  wait, up to that cap. Missing, malformed, or negative values use jitter alone.
- A retry logs a warning with the request method, path, delay, and attempt number. Queries, headers, and payloads
  are omitted from these warnings. Only a final failed call logs an error.

If a retry succeeds, the original call or `async for` continues normally. This applies to reads and streaming
downloads, as well as `POST`, `PUT`, `PATCH`, and `DELETE`: the platform rejects throttled requests before mutation.
Mutating requests are not replayed after ambiguous network failures or HTTP 502, 503, or 504 responses. GET retries
for those server errors share the same attempt and time budget as HTTP 429 retries. Downloads close throttled
responses before waiting, without reading their error bodies.

Assessment preparation and creation reuse each scenario's details within the call. A later call fetches fresh
details. Loading scenario defaults for an atomic assessment uses at most five concurrent requests.

The HTTPX timeout still applies to each network attempt. An attempt already in flight can finish after the retry
budget expires. Use `asyncio.timeout()` around the complete operation when a strict end-to-end deadline is required,
or disable automatic GET rate-limit waiting as shown below.

#### Iterate over every result

Use the normal client and keep processing objects with `async for`. Pass `limit=None` to follow every results page:

```python
import asyncio
import logging

from aiq_platform_api import Assessments, AttackIQClient

logger = logging.getLogger(__name__)

async def main():
    async with AttackIQClient(platform_url, api_token) as client:
        async for result in Assessments.get_results_by_run_id(
            client,
            run_id,
            assessment_version,
            limit=None,
        ):
            logger.info(f"Assessment result: {result}")

asyncio.run(main())
```

Replace the logging statement with the customer's object-processing logic. The same iteration pattern applies to
other paginated SDK reads. The copyable version is in
[`examples/read_rate_limit_handling.py`](https://github.com/AttackIQ/aiq-platform-api/blob/main/examples/read_rate_limit_handling.py)

#### When a 429 is still raised

The raised `httpx.HTTPStatusError` retains the response and its `Retry-After` header. A paginated iterator may already
have yielded earlier objects when a terminal 429 occurs. Logging is harmless; side-effecting processing should be
idempotent. Code that publishes a complete dataset should stage the objects and commit only after iteration finishes.

Worker and scheduler integrations can disable automatic GET rate-limit waiting and handle the first 429 themselves
(this option also covers downloads; mutations still retry platform 429 responses):

```python
import httpx

async with AttackIQClient(
    platform_url,
    api_token,
    retry_rate_limited_gets=False,
) as client:
    try:
        async for result in Assessments.get_results_by_run_id(
            client,
            run_id,
            assessment_version,
            limit=None,
        ):
            logger.info(f"Assessment result: {result}")
    except httpx.HTTPStatusError as error:
        if error.response.status_code != 429:
            raise
        retry_after = error.response.headers.get("Retry-After")
        logger.warning(f"Rate limited; schedule a new collection using Retry-After={retry_after}")
        raise
```

Schedule another run no earlier than a valid returned `Retry-After` delay; use bounded jittered backoff if the header
is absent or invalid. If throttling is frequent, provide the timestamp and the response's `x-aiq-id` to AttackIQ Support
so the tenant policy and request pattern can be reviewed. Never include API tokens or cookies.

---

## Configuration

Both the SDK and CLI require these environment variables:

```sh
export ATTACKIQ_PLATFORM_URL="https://your-platform.attackiq.com"
export ATTACKIQ_PLATFORM_API_TOKEN="your-api-token"
```

Or create a `.env` file in your working directory (auto-loaded).

---

## TLS Verification (on-prem / self-signed certificates)

On-prem servers often present self-signed or non-standards-compliant certificates.
Both the CLI and SDK can skip verification or trust a custom CA bundle.

> **Caveat:** the error `x509: certificate is not standards compliant` is a strict
> certificate-parse rejection, **not** an untrusted-CA error. Only skipping
> verification (`--insecure` / `verify=False`) fixes it — a custom CA bundle
> (`--cacert` / `verify="<path>"`) will **not**.

Shared environment variables (read by both the CLI and the SDK):

```sh
export ATTACKIQ_PLATFORM_INSECURE=true            # skip TLS verification (insecure)
export ATTACKIQ_PLATFORM_CA_BUNDLE=/path/ca.pem   # verify against a custom CA bundle (PEM)
```

`ATTACKIQ_PLATFORM_INSECURE` accepts `1`, `true`, `yes`, or `on` (case-insensitive).
When both vars are set, insecure wins. Disabling verification emits a visible warning.

### CLI

```sh
aiq assets list -k                       # or --insecure; skip verification
aiq assets list --cacert /path/ca.pem    # verify against a custom CA bundle
```

A flag overrides the matching env var (e.g. `--insecure=false` keeps verification on
even when `ATTACKIQ_PLATFORM_INSECURE=true`).

### Python SDK

```python
AttackIQClient(url, token, verify=False)           # skip verification (insecure)
AttackIQClient(url, token, verify="/path/ca.pem")  # verify against a custom CA bundle
```

When `verify` is omitted it falls back to the `ATTACKIQ_PLATFORM_INSECURE` /
`ATTACKIQ_PLATFORM_CA_BUNDLE` env vars; an explicit argument always wins.

---

## CLI

### Quick Install (Recommended)

#### Linux / macOS

```sh
GITHUB_TOKEN="your_token" sh -c 'curl -fsSL -H "Authorization: token $GITHUB_TOKEN" \
  https://raw.githubusercontent.com/AttackIQ/aiq-platform-api/main/install.sh | sh'
```

**Add to PATH** (first time only):
```sh
echo 'export PATH="$HOME/.local/bin:$PATH"' >> ~/.zshrc  # or ~/.bashrc
```

Auto-detects OS/arch, installs to `~/.local/bin` (no sudo).

#### Windows (Native)

**PowerShell installer:**
```powershell
$env:GITHUB_TOKEN = "your_token"
Invoke-WebRequest -Uri "https://raw.githubusercontent.com/AttackIQ/aiq-platform-api/main/install.ps1" -Headers @{Authorization="token $env:GITHUB_TOKEN"} -OutFile "$env:TEMP\install.ps1"
powershell -ExecutionPolicy Bypass -File "$env:TEMP\install.ps1"
```

Installs to `%LOCALAPPDATA%\Programs\aiq` and adds to PATH automatically.

### Usage

```sh
# List available commands
aiq --help

# List assessments
aiq assessments list

# Search assets
aiq assets search --query "hostname"

# Get scenario details
aiq scenarios get --scenario-id "abc123"
```

### Shell Completion

The CLI supports shell completion for bash, zsh, fish, and PowerShell.

#### Bash

**Current session:**
```sh
source <(aiq completion bash)
```

**Permanent installation:**
```sh
# Linux
aiq completion bash | sudo tee /etc/bash_completion.d/aiq

# macOS
aiq completion bash > $(brew --prefix)/etc/bash_completion.d/aiq
```

#### Zsh

**Current session:**
```sh
source <(aiq completion zsh)
```

**Permanent installation:**
```sh
# Add to ~/.zshrc
echo "source <(aiq completion zsh)" >> ~/.zshrc

# Or install to completions directory
aiq completion zsh > "${fpath[1]}/_aiq"
```

#### Fish

**Permanent installation:**
```sh
aiq completion fish | source

# Or save to completions directory
aiq completion fish > ~/.config/fish/completions/aiq.fish
```

#### PowerShell

**Current session:**
```powershell
aiq completion powershell | Out-String | Invoke-Expression
```

**Permanent installation:**
Add the following to your PowerShell profile:
```powershell
aiq completion powershell | Out-String | Invoke-Expression
```

## Cross-surface assessment planning

The Python SDK and Go CLI provide additive planning and creation APIs. Existing
`create_atomic_assessment` and `aiq assessments create-atomic` remain available.

```sh
# Read-only preflight: discovers compatible active agents when asset IDs are omitted.
aiq assessments prepare --scenario-id SCENARIO_ID

# Read-only coordinated plan; supplied asset IDs form the candidate inventory.
aiq assessments plan-cross-surface --scenario-id WINDOWS_SCENARIO,LINUX_SCENARIO --asset-id WINDOWS_AGENT,LINUX_AGENT

# Create one surface/method group, preserving its planned targets.
aiq assessments create --name "Validation" --scenario-id SCENARIO_ID --asset-id AGENT_ID --target-attack-surface endpoint --test-method atomic
```

All three commands return JSON. `prepare` and `plan-cross-surface` exit nonzero
when preflight fails. `--scenario-configurations configs.json` supplies an object
mapping scenario IDs to their `model_json` overrides; `--asset-tag-id` supplies
required asset tags. `prepare` and `create` also accept `--use-hosted-agent`.
Explicit targets are validated rather than replaced. The planner covers every
scenario's supported runtime/platform and minimum target count; unsupported
connector or unspecified non-endpoint runtimes fail before creation.

For each valid group, pass its scenario IDs, surface, method, selected agent IDs
and hosted flag to `create`, then use the existing `run` and `status` commands.
Planning does not execute assessments. Live validation is available through
`aiq test plan-cross-surface --scenario-id SCENARIO_ID [--asset-id AGENT_ID]`.

## Contributing

We welcome feedback and contributions! For detailed contribution guidelines, please see [CONTRIBUTING.md](CONTRIBUTING.md).

Quick ways to contribute:
- Open issues for bugs or feature requests
- Submit pull requests
- Provide feedback on the API design

## License

MIT License - See LICENSE file for details
