Examples & Scripts¶

Focused demos and operational scripts for the Silo SDK. Each demo makes real API calls to demonstrate module capabilities. Scripts provide automation for user management, session reports, log extraction, and file operations.

Prerequisites¶

source .venv/bin/activate
pip install -e .
source .env   # or export tokens manually

Quick Reference¶

File	API Module	Token Required
`browsing_demo.py`	BrowsingAPI	`A8_ADMIN_TOKEN`
`user_demo.py`	UserManagementAPI	`A8_SYNC_TOKEN`
`org_demo.py`	OrgManagementAPI	`A8_ADMIN_TOKEN`
`file_demo.py`	FileAPI	`A8_FILE_TOKEN`
`log_demo.py`	LogExtractionAPI	`A8_LOG_TOKEN`
`harvest_demo_sync.py`	HarvesterAPI (sync)	`A8_SCRAPE_TOKEN`
`harvest_demo_async.py`	HarvesterAPI (async)	`A8_SCRAPE_TOKEN`
`config_demo.py`	Configuration	(none)

Each demo:

Checks for its required token at startup and skips gracefully if missing
Loads config from config/default.json with ${A8_*} env var substitution
Supports --help for usage info
Makes real API calls with proper exception handling

Browsing Isolation¶

Full demo for BrowsingAPI — creating contexts, building launch URLs, and getting context status.

Requires: A8_ADMIN_TOKEN

examples/browsing_demo.py

#!/usr/bin/env python3
r"""
Isolation API Demo.
==================
This script demonstrates how to use the BrowsingAPI to create secure
browsing contexts for multiple URLs and retrieve launch URLs.

Features:
- Pass multiple URLs to create isolated browsing contexts
- Apply customizable security policies via CLI flags
- Support for Silo for Research with egress regions and browser profiles
- Get back launch URLs for each context
- Optionally open the generated URLs in a browser

Usage:
    # Basic usage with default URLs
    python3 browsing_demo.py

    # With custom URLs
    python3 browsing_demo.py https://example.com https://github.com

    # Specify a user
    python3 browsing_demo.py --user analyst@company.com https://example.com

    # Open links in browser automatically
    python3 browsing_demo.py --open-links https://example.com

    # Silo for Research with egress location
    python3 browsing_demo.py --research --egress "New York, NY" https://example.com

    # With browser profile
    python3 browsing_demo.py --research --egress London \
        --browser chrome --os win https://example.com

    # List available egress locations
    python3 browsing_demo.py --list-egress

    # With security policies
    python3 browsing_demo.py --readonly --block-clipboard --block-files https://example.com

    # With UI customization
    python3 browsing_demo.py --minimal-ui --ribbon-message "Secure Session" https://example.com

    # Full example with multiple options
    python3 browsing_demo.py \\
        --user analyst@company.com \\
        --research \\
        --egress "New York, NY" \\
        --browser chrome --os win \\
        --readonly --block-files --block-clipboard \\
        --minimal-ui \\
        --ribbon-message "Research Session" \\
        --ribbon-bg-color "#0000FF" \\
        --open-links \\
        https://example.com

CLI Options:
    Positional:
        input_urls              URLs to create isolation contexts for

    Session Options:
        --user USERNAME         Username for the session
        --research              Create Silo for Research session
        --egress REGION         Egress region (e.g., 'New York, NY', 'London')
        --os OS                 OS for browser profile (win, mac, linux, android, ios)
        --browser BROWSER       Browser type (chrome, firefox, edge, safari, tor)

    Security Policies:
        --readonly              Enable read-only mode (block form input)
        --block-clipboard       Block all clipboard operations
        --block-files           Block all file transfers
        --block-print           Block printing
        --block-ads             Enable ad blocking

    Domain/Category Filtering:
        --allow-domains DOMAINS     Comma-separated domains to allow
        --block-domains DOMAINS     Comma-separated domains to block
        --block-categories CATS     Comma-separated categories to block

    UI Customization:
        --minimal-ui            Use minimal UI (ribbon only, no URL bar)
        --seamless-ui           Use seamless UI (no ribbon or URL bar)
        --ribbon-message TEXT   Custom ribbon message (max 100 chars)
        --ribbon-bg-color HEX   Ribbon background color (e.g., '#0000FF')
        --ribbon-text-color HEX Ribbon text color (e.g., '#FFFFFF')

    Other:
        --open-links            Open generated URLs in browser
        --list-egress           List available egress locations
        --policy-json JSON      Full policy as JSON (overrides other policy flags)

Requirements:
    - ADMIN_TOKEN must be set in config or environment
    - Optional: CATCHALL_USER for default username

"""

import argparse
import json
import sys
import time
import webbrowser
from typing import Any, cast

from silo_sdk import EGRESS_LOCATIONS, BrowsingAPI, load_config
from silo_sdk.base.exceptions import (
    BrowsingAPIError,
    ConfigurationError,
    ValidationError,
)


def get_default_policy(config: dict[str, Any] | None = None) -> list[dict[str, Any]]:
    """
    Return the default security policy for browsing contexts.

    Reads from config/default.json. Returns empty list if not configured,
    allowing the user's default policy to be applied.

    Args:
        config: Optional config dictionary. If None, will load from config files.

    Returns:
        List of policy dictionaries defining session behavior.
    """
    if config and "DEFAULT_POLICY" in config:
        policy = config["DEFAULT_POLICY"]
        if isinstance(policy, list):
            return policy

    # Return empty list - user's default policy will be applied
    return []


def create_isolation_urls(
    input_urls: list[str],
    username: str | None = None,
    policy: list[dict[str, Any]] | None = None,
    max_uses: int = 1,
    egress_region: str | None = None,
    browser_profile: dict[str, str] | None = None,
    research_mode: bool = False,
) -> dict[str, Any]:
    """
    Create secure browsing contexts for multiple URLs and return launch URLs.

    Args:
        input_urls: List of URLs to create contexts for.
        username: Username associated with contexts. Uses config default if None.
        policy: Security policy list. Uses default policy if None.
        max_uses: Maximum number of times each context can be used.
        egress_region: Egress location for Silo for Research.
        browser_profile: Browser profile dict with 'os'/'browser' keys (and
            optional 'timezone'/'languages').
        research_mode: If True, creates Silo for Research sessions.

    Returns:
        Dictionary with 'context_urls' key containing list of launch URLs.

    Raises:
        ConfigurationError: If required configuration is missing.
        BrowsingAPIError: If context creation fails.
    """
    # Load configuration
    config = load_config()

    # Initialize the BrowsingAPI
    browsing = BrowsingAPI(config)

    # Use default username from config if not provided
    if username is None:
        username = config.get("CATCHALL_USER", "demo_user")

    # Use default policy from config if not provided
    if policy is None:
        policy = get_default_policy(config)

    # Determine session type
    session_type: str | None = "toolbox-standalone" if research_mode else None

    # Build categories if using research mode with egress
    categories: dict[str, str] | None = None
    if research_mode and egress_region:
        categories = {"availability": "public", "connectivity": "datacenter"}

    # Create contexts for all URLs
    result = browsing.bulk_create_contexts(
        urls=input_urls,
        username=username,
        policy=policy if policy else None,
        max_uses=max_uses,
        session_type=cast(Any, session_type),
        egress_region=egress_region,
        browser_profile=cast(Any, browser_profile),
        categories=cast(Any, categories),
    )

    return result


def build_policy_from_args(
    readonly: bool = False,
    block_clipboard: bool = False,
    block_files: bool = False,
    minimal_ui: bool = False,
    seamless_ui: bool = False,
    ribbon_message: str | None = None,
    ribbon_bg_color: str | None = None,
    ribbon_text_color: str | None = None,
    block_print: bool = False,
    block_ads: bool = False,
    allow_domains: str | None = None,
    block_domains: str | None = None,
    block_categories: str | None = None,
    policy_json: str | None = None,
) -> list[dict[str, Any]] | None:
    """
    Build a policy list from CLI arguments.

    Args:
        readonly: Enable read-only mode
        block_clipboard: Block all clipboard operations
        block_files: Block all file transfers
        minimal_ui: Use minimal browser chrome (ribbon only)
        seamless_ui: Use seamless browser chrome (no ribbon or URL bar)
        ribbon_message: Custom ribbon message
        ribbon_bg_color: Ribbon background color (hex)
        ribbon_text_color: Ribbon text color (hex)
        block_print: Block printing
        block_ads: Enable ad blocking
        allow_domains: Comma-separated domains to allow
        block_domains: Comma-separated domains to block
        block_categories: Comma-separated categories to block
        policy_json: Full policy as JSON string (overrides other flags)

    Returns:
        List of policy dictionaries, or None if no policy specified.
    """
    # If JSON policy provided, parse and return it
    if policy_json:
        try:
            parsed = json.loads(policy_json)
            if isinstance(parsed, list):
                return parsed
            print("Error: --policy-json must be a JSON array")
            sys.exit(1)
        except json.JSONDecodeError as e:
            print(f"Error parsing --policy-json: {e}")
            sys.exit(1)

    # Build policy from individual flags
    policy = []

    if readonly:
        policy.append({"type": "readonly", "params": ["true"]})

    if block_clipboard:
        policy.append({"type": "clipboard", "params": ["block_all"]})

    if block_files:
        policy.append({"type": "file_transfer", "params": ["block_all"]})

    # Browser chrome - seamless takes priority over minimal
    if seamless_ui:
        policy.append({"type": "browser_chrome", "params": ["seamless"]})
    elif minimal_ui:
        policy.append({"type": "browser_chrome", "params": ["minimal"]})

    if ribbon_message:
        policy.append({"type": "ribbon_message", "params": [ribbon_message[:100]]})

    if ribbon_bg_color:
        policy.append({"type": "ribbon_background_color", "params": [ribbon_bg_color]})

    if ribbon_text_color:
        policy.append({"type": "ribbon_text_color", "params": [ribbon_text_color]})

    if block_print:
        policy.append({"type": "print", "params": ["block"]})

    if block_ads:
        policy.append({"type": "ad_block", "params": ["enable"]})

    if allow_domains:
        domains = [d.strip() for d in allow_domains.split(",")]
        policy.append({"type": "domain_allow", "params": domains})

    if block_domains:
        domains = [d.strip() for d in block_domains.split(",")]
        policy.append({"type": "domain_block", "params": domains})

    if block_categories:
        categories = [c.strip() for c in block_categories.split(",")]
        policy.append({"type": "category_block", "params": categories})

    return policy if policy else None


def list_egress_locations() -> None:
    """Print available egress locations by region."""
    print("Available Egress Locations:")
    print("=" * 50)
    for region, locations in EGRESS_LOCATIONS.items():
        print(f"\n{region}:")
        for loc in locations:
            print(f"  - {loc}")


def main(
    input_urls: list[str] | None = None,
    open_links: bool = False,
    username: str | None = None,
    research_mode: bool = False,
    egress_region: str | None = None,
    browser_os: str | None = None,
    browser_type: str | None = None,
    policy: list[dict[str, Any]] | None = None,
) -> None:
    """
    Main function to process URLs and create secure browsing contexts.

    Args:
        input_urls: List of URLs to process. Uses defaults if None.
        open_links: Whether to automatically open generated URLs in browser.
        username: Username for the session. Uses CATCHALL_USER if None.
        research_mode: If True, creates Silo for Research sessions.
        egress_region: Egress location for research sessions.
        browser_os: OS for browser profile (win, mac, linux, android, ios).
        browser_type: Browser type (chrome, firefox, edge, safari, tor).
        policy: Custom policy list. Uses config default if None.
    """
    # Default URLs for demo if none provided
    if not input_urls:
        input_urls = [
            "https://www.example.com",
            "https://github.com",
            "https://docs.python.org",
        ]
        print("No URLs provided. Using demo URLs:")
        for url in input_urls:
            print(f"  - {url}")
        print()

    # Build browser profile if specified
    browser_profile = None
    if browser_os or browser_type:
        browser_profile = {}
        if browser_os:
            browser_profile["os"] = browser_os
        if browser_type:
            browser_profile["browser"] = browser_type

    try:
        mode_str = "Silo for Research" if research_mode else "isolation"
        print(f"Creating {mode_str} contexts for {len(input_urls)} URL(s)...")
        if research_mode and egress_region:
            print(f"Egress region: {egress_region}")
        if browser_profile:
            print(f"Browser profile: {browser_profile}")
        print("-" * 50)

        result = create_isolation_urls(
            input_urls,
            username=username,
            policy=policy,
            egress_region=egress_region,
            browser_profile=browser_profile,
            research_mode=research_mode,
        )

        # Print results
        print("\nGenerated Launch URLs:")
        print("=" * 50)
        for i, launch_url in enumerate(result["context_urls"], 1):
            print(f"{i}. {launch_url}")

        print(f"\nSuccessfully created {len(result['context_urls'])} context(s)")

        # Print full JSON result
        print("\nFull Result (JSON):")
        print(json.dumps(result, indent=2))

        # Optionally open links in browser
        if open_links and result["context_urls"]:
            print("\nOpening links in browser...")
            for launch_url in result["context_urls"]:
                print(f"Opening: {launch_url}")
                webbrowser.open(launch_url)
                time.sleep(3)  # Brief pause between opening tabs

    except ConfigurationError as e:
        print(f"Configuration Error: {e}")
        print("\nMake sure ADMIN_TOKEN is set in your config or environment.")
        sys.exit(1)
    except ValidationError as e:
        print(f"Validation Error: {e}")
        sys.exit(1)
    except BrowsingAPIError as e:
        print(f"API Error: {e}")
        sys.exit(1)
    except Exception as e:
        print(f"Unexpected Error: {e}")
        sys.exit(1)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="Create secure browsing isolation contexts for multiple URLs",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  python3 browsing_demo.py
      Create contexts for default demo URLs

  python3 browsing_demo.py https://example.com https://github.com
      Create contexts for specified URLs

  python3 browsing_demo.py --user analyst@company.com https://example.com
      Create context for specific user

  python3 browsing_demo.py --research --egress "New York, NY" https://example.com
      Create Silo for Research session with US egress

  python3 browsing_demo.py --readonly --block-files --minimal-ui https://example.com
      Create context with security policies

  python3 browsing_demo.py --list-egress
      List all available egress locations

Available categories for --block-categories:
  Questionable and Offensive, Adult and Pornographic, Malicious Sites,
  Social Networking, Shopping, Rich Media, Job Search,
  Network Hogs & Data Leaks, Finance, Health
        """,
    )

    # Positional arguments
    parser.add_argument(
        "input_urls",
        nargs="*",
        help="URLs to create isolation contexts for (defanged URLs supported)",
    )

    # Session options group
    session_group = parser.add_argument_group("Session Options")
    session_group.add_argument(
        "--user",
        type=str,
        metavar="USERNAME",
        help="Username for the session (defaults to CATCHALL_USER from config)",
    )
    session_group.add_argument(
        "--research",
        action="store_true",
        help="Create Silo for Research (toolbox-standalone) session",
    )
    session_group.add_argument(
        "--egress",
        type=str,
        metavar="REGION",
        help=(
            "Egress region (e.g., 'New York, NY', 'London'). "
            "Use --list-egress to see options"
        ),
    )
    session_group.add_argument(
        "--os",
        type=str,
        choices=["win", "mac", "linux", "android", "ios"],
        help="OS for browser profile",
    )
    session_group.add_argument(
        "--browser",
        type=str,
        choices=["chrome", "firefox", "edge", "safari", "tor"],
        help="Browser type for browser profile",
    )

    # Security policies group
    security_group = parser.add_argument_group("Security Policies")
    security_group.add_argument(
        "--readonly",
        action="store_true",
        help="Enable read-only mode (block form input)",
    )
    security_group.add_argument(
        "--block-clipboard",
        action="store_true",
        help="Block all clipboard operations",
    )
    security_group.add_argument(
        "--block-files",
        action="store_true",
        help="Block all file transfers",
    )
    security_group.add_argument(
        "--block-print",
        action="store_true",
        help="Block printing",
    )
    security_group.add_argument(
        "--block-ads",
        action="store_true",
        help="Enable ad blocking",
    )

    # Domain/category filtering group
    filter_group = parser.add_argument_group("Domain & Category Filtering")
    filter_group.add_argument(
        "--allow-domains",
        type=str,
        metavar="DOMAINS",
        help="Comma-separated domains to allow (blocks all others)",
    )
    filter_group.add_argument(
        "--block-domains",
        type=str,
        metavar="DOMAINS",
        help="Comma-separated domains to block (allows all others)",
    )
    filter_group.add_argument(
        "--block-categories",
        type=str,
        metavar="CATEGORIES",
        help="Comma-separated categories to block",
    )

    # UI customization group
    ui_group = parser.add_argument_group("UI Customization")
    ui_group.add_argument(
        "--minimal-ui",
        action="store_true",
        help="Minimal UI (ribbon only, no URL bar)",
    )
    ui_group.add_argument(
        "--seamless-ui",
        action="store_true",
        help="Seamless UI (no ribbon or URL bar)",
    )
    ui_group.add_argument(
        "--ribbon-message",
        type=str,
        metavar="TEXT",
        help="Custom ribbon message (max 100 chars)",
    )
    ui_group.add_argument(
        "--ribbon-bg-color",
        type=str,
        metavar="HEX",
        help="Ribbon background color (e.g., '#0000FF')",
    )
    ui_group.add_argument(
        "--ribbon-text-color",
        type=str,
        metavar="HEX",
        help="Ribbon text color (e.g., '#FFFFFF')",
    )

    # Other options group
    other_group = parser.add_argument_group("Other Options")
    other_group.add_argument(
        "--open-links",
        action="store_true",
        help="Open generated URLs in browser",
    )
    other_group.add_argument(
        "--list-egress",
        action="store_true",
        help="List available egress locations and exit",
    )
    other_group.add_argument(
        "--policy-json",
        type=str,
        metavar="JSON",
        help="Full policy as JSON (overrides other policy flags)",
    )

    args = parser.parse_args()

    # Handle --list-egress
    if args.list_egress:
        list_egress_locations()
        sys.exit(0)

    # Build policy from CLI arguments
    policy = build_policy_from_args(
        readonly=args.readonly,
        block_clipboard=args.block_clipboard,
        block_files=args.block_files,
        minimal_ui=args.minimal_ui,
        seamless_ui=args.seamless_ui,
        ribbon_message=args.ribbon_message,
        ribbon_bg_color=args.ribbon_bg_color,
        ribbon_text_color=args.ribbon_text_color,
        block_print=args.block_print,
        block_ads=args.block_ads,
        allow_domains=args.allow_domains,
        block_domains=args.block_domains,
        block_categories=args.block_categories,
        policy_json=args.policy_json,
    )

    main(
        input_urls=args.input_urls if args.input_urls else None,
        open_links=args.open_links,
        username=args.user,
        research_mode=args.research,
        egress_region=args.egress,
        browser_os=args.os,
        browser_type=args.browser,
        policy=policy,
    )

Usage:

python examples/browsing_demo.py https://example.com

File Storage¶

FileAPI walkthrough — upload, search, download, and delete. Includes async bulk operations.

Requires: A8_FILE_TOKEN

examples/file_demo.py

#!/usr/bin/env python3
"""
File API Demo.
=============
Demonstrates file upload, search, modification, download, and deletion
using the FileAPI.

Usage:
    python examples/file_demo.py

Requirements:
    - A8_FILE_TOKEN must be set in config or environment
    - A8_BUCKET_ID must be set in config or environment

Modified: 2025-02-10
"""

import os
import secrets
import sys
import tempfile
import time
from pathlib import Path

from silo_sdk import FileAPI, load_config
from silo_sdk.base.exceptions import ConfigurationError, FileAPIError, ValidationError


def main() -> None:
    if "--help" in sys.argv or "-h" in sys.argv:
        print(__doc__)
        sys.exit(0)

    print("File API Demo")
    print("=" * 60)

    # Load configuration
    config = load_config("config/default.json")

    # Check required tokens
    if not config.get("FILE_TOKEN"):
        print("Skipping: FILE_TOKEN not configured.")
        print("Set A8_FILE_TOKEN in your .env or config/default.json")
        sys.exit(0)

    bucket_id = config.get("BUCKET_ID")
    if not bucket_id:
        print("Skipping: BUCKET_ID not configured.")
        print("Set A8_BUCKET_ID in your .env or config/default.json")
        sys.exit(0)

    try:
        files = FileAPI(config)
        print(f"FileAPI initialized (bucket: {bucket_id})")

        # --- Upload ---
        print("\n--- Upload ---")
        random_id = str(1000 + secrets.randbelow(9000))
        local_file = f"demo_file_{random_id}.txt"

        timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
        with open(local_file, "w", encoding="utf-8") as f:
            f.write(f"File API Demo - {timestamp}\n")
            f.write(f"Random ID: {random_id}\n")
            f.write("This file tests upload, search, modify, download, and delete.\n")

        file_size = os.path.getsize(local_file)
        print(f"Created test file: {local_file} ({file_size} bytes)")

        try:
            upload_result = files.upload_file(
                bucket_id=bucket_id, file_path=local_file, name=local_file
            )
            file_id = upload_result["file_id"]
            print(f"Uploaded. File ID: {file_id}")
        finally:
            if os.path.exists(local_file):
                os.unlink(local_file)

        # --- Search ---
        print("\n--- Search ---")
        found = files.find_files(bucket_id=bucket_id, name=f"demo_file_{random_id}*")
        print(f"Found {len(found)} file(s) matching 'demo_file_{random_id}*'")
        for f_info in found:
            print(f"  {f_info.get('name')} (ID: {f_info.get('file_id')})")

        # --- Get Info ---
        print("\n--- File Info ---")
        details = files.get_file_info(file_id)
        print(f"  Name:         {details.get('name')}")
        print(f"  Size:         {details.get('size', 'Unknown')} bytes")
        print(f"  Content-Type: {details.get('content_type', 'Unknown')}")
        print(f"  Created:      {details.get('created_at', 'Unknown')}")

        # --- Modify ---
        print("\n--- Modify ---")
        new_name = f"renamed_demo_{random_id}.txt"
        files.modify_file(file_id=file_id, name=new_name, content_type="text/plain")
        print(f"Renamed to: {new_name}")

        # --- Download ---
        print("\n--- Download ---")
        download_dir = Path(tempfile.gettempdir()) / "file_api_demo"
        download_dir.mkdir(exist_ok=True)
        output_path = download_dir / new_name

        downloaded = files.download_file(file_id=file_id, output_path=output_path)
        dl_size = os.path.getsize(downloaded)
        print(f"Downloaded to: {downloaded} ({dl_size} bytes)")

        with open(downloaded, encoding="utf-8") as f:
            lines = f.readlines()
            print(f"  Content: {len(lines)} lines")
            for line in lines[:3]:
                print(f"    {line.rstrip()}")

        os.unlink(downloaded)
        try:
            download_dir.rmdir()
        except OSError:
            pass

        # --- List ---
        print("\n--- List Files ---")
        recent = files.list_files(bucket_id=bucket_id, limit=5)
        print(f"Recent files in bucket ({len(recent)} shown):")
        for f_info in recent:
            print(
                f"  {f_info.get('name', 'Unknown'):40s}"
                f"  {f_info.get('size', '?')} bytes"
            )

        # --- Delete ---
        print("\n--- Delete ---")
        files.delete_file(file_id=file_id)
        print(f"Deleted file: {file_id}")

        # --- Summary ---
        print("\n" + "=" * 60)
        print("File API Demo Summary:")
        print("  Upload, search, info, modify, download, list, delete - all OK")

    except ConfigurationError as e:
        print(f"Configuration Error: {e}")
        sys.exit(1)
    except ValidationError as e:
        print(f"Validation Error: {e}")
        sys.exit(1)
    except FileAPIError as e:
        print(f"File API Error: {e}")
        sys.exit(1)


if __name__ == "__main__":
    main()

Usage:

python examples/file_demo.py

Log Extraction & Decryption¶

Log Extraction Demo¶

LogExtractionAPI walkthrough — log types, sequence info, extraction, and decryption workflow.

Requires: A8_LOG_TOKEN

examples/log_demo.py

#!/usr/bin/env python3
"""
Log Extraction Demo.
===================
Demonstrates retrieving available log types, checking sequence info,
extracting a small sample of logs using the LogExtractionAPI, and
the full extract → decrypt workflow using native SDK decryption.

Usage:
    python examples/log_demo.py
    python examples/log_demo.py --pvtkey /path/to/pvtkey.txt

Requirements:
    - A8_LOG_TOKEN must be set in config or environment
    - A8_TOP_ORG environment variable set to the organization name
    - Optional: --pvtkey for the decryption workflow section
    - Optional: pip install "silo-sdk[decrypt]" for Standard decryption

Modified: 2026-02-27
"""

import os
import sys

from silo_sdk import LogExtractionAPI, load_config
from silo_sdk.base.exceptions import (
    ConfigurationError,
    LogExtractionAPIError,
    ValidationError,
)


def main() -> None:
    import argparse

    parser = argparse.ArgumentParser(
        description="Log Extraction Demo",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
    )
    parser.add_argument(
        "--pvtkey",
        type=str,
        metavar="PATH",
        default=None,
        help="Path to pvtkey.txt for the decryption workflow section (optional)",
    )
    args = parser.parse_args()

    print("Log Extraction Demo")
    print("=" * 60)

    # Load configuration
    config = load_config("config/default.json")

    # Check required tokens
    if not config.get("LOG_TOKEN"):
        print("Skipping: LOG_TOKEN not configured.")
        print("Set A8_LOG_TOKEN in your .env or config/default.json")
        sys.exit(0)

    org = os.environ.get("A8_TOP_ORG")
    if not org:
        print("Skipping: A8_TOP_ORG environment variable not set.")
        sys.exit(0)

    try:
        logs = LogExtractionAPI(config)
        print(f"LogExtractionAPI initialized (org: {org})")

        # --- Valid Log Types ---
        print("\n--- Valid Log Types ---")
        log_types = LogExtractionAPI.get_valid_log_types()
        for i, lt in enumerate(sorted(log_types), 1):
            print(f"  {i:2d}. {lt}")
        print(f"\n  {len(log_types)} log types supported")

        # Highlight additional log types
        new_types = [
            "EXPLOIT",
            "PRINT",
            "SMS",
            "ISOLATE_BYPASS",
            "TRAFFICMAN",
            "HARVEST",
            "CASE_MANAGER",
            "CLIPBOARD",
            "LAUNCHER",
        ]
        print("\n  Additional log types:")
        for lt in sorted(new_types):
            if lt not in log_types:
                print(f"  WARNING: Missing expected log type: {lt}")
            print(f"    - {lt}")

        # --- Sequence Info ---
        print(f"\n--- Log Sequence Info for '{org}' ---")
        seq_info = logs.get_log_sequence_info(org)
        min_seq = seq_info.get("min_seq", "N/A")
        max_seq = seq_info.get("max_seq", "N/A")
        print(f"  Min sequence: {min_seq}")
        print(f"  Max sequence: {max_seq}")

        # --- Extract Sample Logs ---
        print("\n--- Extract Sample Logs ---")
        sample_types = ["AUTH", "SESSION"]
        start_seq = 0

        # Use min_seq if available
        if isinstance(min_seq, int):
            start_seq = min_seq

        result = logs.extract_logs(
            org=org,
            start_seq=start_seq,
            log_types=sample_types,
            limit=5,
        )

        # --- Date-Based Filtering Example ---
        print("\n--- Date-Based Filtering (last 7 days) ---")
        print("  Example call: extract_logs(org, 0, ['AUTH'], start_date=7)")
        print("  start_date=7 means 'from 7 days ago'")
        print("  end_date=0   means 'until today'")
        print("  Also accepts datetime objects for precise control")

        entries = result.get("logs", result.get("entries", []))
        print(f"Requested types: {sample_types}")
        print(f"Start sequence:  {start_seq}")
        print(f"Entries returned: {len(entries)}")

        if entries:
            print("\nSample entries:")
            for entry in entries[:5]:
                timestamp = entry.get("timestamp", entry.get("ts", ""))
                log_type = entry.get("log_type", entry.get("type", ""))
                username = entry.get("username", entry.get("user", ""))
                print(f"  [{timestamp}] {log_type:12s}  {username}")

        # --- Decryption Workflow ---
        print("\n--- Decryption Workflow (ENC Log Type) ---")
        print("  ENC log entries are encrypted at rest and require decryption.")
        print("  The SDK provides native decryption.")
        print()
        print("  Install the decrypt extra:")
        print('    pip install "silo-sdk[decrypt]"         # Standard (EC+HKDF)')
        print('    pip install "silo-sdk[legacy-decrypt]"  # Legacy (seccure)')
        print()
        print("  Full extract → decrypt workflow:")
        print(
            "    from silo_sdk.logging.decrypt import decrypt_logs, load_private_keys"
        )
        print()
        print("    # 1. Extract ENC logs from the API")
        print(
            "    enc_logs = logs.extract_all_logs(org, start_seq=0, log_types=['ENC'])"
        )
        print()
        print("    # 2. Load your private key(s)")
        print("    keys = load_private_keys('pvtkey.txt', load_pem=True)")
        print()
        print("    # 3. Decrypt — merges plaintext fields into each entry dict")
        print("    decrypted = decrypt_logs(enc_logs, keys)")
        print()
        print("    # 4. Use decrypted fields normally")
        print("    for entry in decrypted:")
        print("        print(entry.get('url'), entry.get('username'))")
        print()

        if args.pvtkey:
            from silo_sdk.logging.decrypt import decrypt_logs, load_private_keys

            print(f"  Loading keys from: {args.pvtkey}")
            try:
                keys = load_private_keys(args.pvtkey, load_pem=True)
                print(f"  Loaded {len(keys)} key(s): {list(keys.keys())}")

                # Filter for ENC entries from sample
                enc_entries = [e for e in entries if e.get("type") == "ENC"]
                if enc_entries:
                    print(f"  Decrypting {len(enc_entries)} ENC sample entries...")
                    decrypted = decrypt_logs(enc_entries, keys)
                    print(
                        f"  Decrypted: {len(decrypted)} of {len(enc_entries)} entries"
                    )
                else:
                    print("  No ENC entries in the sample batch.")
                    print("  Extract ENC-only logs to see decryption in action:")
                    print("    enc_logs = logs.extract_all_logs(org, 0, ['ENC'])")
                    decrypted = []
            except FileNotFoundError:
                print(f"  Key file not found: {args.pvtkey}")
                decrypted = []
        else:
            print("  Run with --pvtkey /path/to/pvtkey.txt to demo live decryption.")
            print("  See examples/decrypt_demo.py for a full decryption walkthrough.")

        # --- Summary ---
        print("\n" + "=" * 60)
        print("Log Extraction Demo Summary:")
        print(f"  Organization:   {org}")
        print(f"  Sequence range: {min_seq} - {max_seq}")
        print(f"  Sample entries: {len(entries)}")
        print(f"  Log types:      {len(log_types)} supported")

    except ConfigurationError as e:
        print(f"Configuration Error: {e}")
        sys.exit(1)
    except ValidationError as e:
        print(f"Validation Error: {e}")
        sys.exit(1)
    except LogExtractionAPIError as e:
        print(f"Log Extraction API Error: {e}")
        sys.exit(1)


if __name__ == "__main__":
    main()

Usage:

python examples/log_demo.py

Decryption Demo¶

Full decryption API walkthrough — loading keys, decrypting log entries and video files.

Requires: pip install -e ".[decrypt]" (from SDK directory) and a pvtkey.txt file.

examples/decrypt_demo.py

#!/usr/bin/env python3
r"""
Log Decryption Demo.
===================
Demonstrates decrypting encrypted Silo platform log entries using the
native decryption functions in ``silo_sdk.logging.decrypt``.

Covers:
    - pvtkey.txt file format and loading with ``load_private_keys()``
    - Single-entry decryption with ``decrypt_log_entry()``
    - Batch decryption with ``decrypt_logs()``
    - Video file decryption with ``decrypt_video_file()``
    - Standard (EC + HKDF-SHA384 + AES-256-GCM) and Legacy (seccure) paths

Usage:
    python examples/decrypt_demo.py
    python examples/decrypt_demo.py --pvtkey /path/to/pvtkey.txt
    python examples/decrypt_demo.py --pvtkey pvtkey.txt --video /path/to/video.enc

Requirements:
    - ``pip install "silo-sdk[decrypt]"`` for Standard decryption
    - ``pip install "silo-sdk[legacy-decrypt]"`` for Legacy decryption
    - A pvtkey.txt file with at least one key (optional — demo runs without it)

pvtkey.txt format:
    Each line is:  key-name=key-value
    For Standard:  mykey=-----BEGIN EC PRIVATE KEY-----\\n...\\n-----END EC PRIVATE KEY-----
    For Legacy:    legacy_key=raw_passphrase_value
    PEM files:     mykey=/path/to/private.pem   (use --load-pem to dereference)

Modified: 2026-02-27
"""

import argparse
from typing import Any, cast

from silo_sdk.logging.decrypt import (
    decrypt_log_entry,
    decrypt_logs,
    decrypt_video_file,
    load_private_keys,
)

# ---------------------------------------------------------------------------
# Sample encrypted entries — these represent real API output format.
# The `enc` field is base64-encoded ciphertext; `key_name` identifies
# which private key to use; `encryption_type` is "Standard" or "Legacy".
# ---------------------------------------------------------------------------
SAMPLE_ENC_ENTRIES = [
    {
        "create_ts": "2026-01-15T10:30:00Z",
        "type": "ENC",
        "seq_id": 1001,
        "key_name": "mykey",
        "encryption_type": "Standard",
        "enc": "BASE64_ENCODED_CIPHERTEXT_HERE",
    },
    {
        "create_ts": "2026-01-15T10:31:00Z",
        "type": "ENC",
        "seq_id": 1002,
        "key_name": "mykey",
        "encryption_type": "Standard",
        "enc": "BASE64_ENCODED_CIPHERTEXT_HERE",
    },
    {
        "create_ts": "2026-01-15T10:32:00Z",
        "type": "ENC",
        "seq_id": 1003,
        "key_name": "legacy_key",
        "encryption_type": "Legacy",
        "enc": "BASE64_ENCODED_CIPHERTEXT_HERE",
    },
]


def show_key_file_format() -> None:
    """Print pvtkey.txt format documentation."""
    print("\n--- pvtkey.txt Format ---")
    print("  Each non-empty line: key-name=key-value")
    print()
    print("  Standard encryption (EC private key inline):")
    print(
        "    mykey=-----BEGIN EC PRIVATE KEY-----\\n"
        "MHQCAQEEIBn...\\n-----END EC PRIVATE KEY-----"
    )
    print()
    print("  Standard encryption (reference a .pem file):")
    print("    mykey=/path/to/private.pem")
    print()
    print("  Legacy encryption (raw passphrase):")
    print("    legacy_key=my_secret_passphrase")
    print()
    print("  Manage keys with: python scripts/key_manager.py --help")


def demo_load_keys(pvtkey_path: str) -> dict[str, str]:
    """Demonstrate load_private_keys()."""
    print("\n--- Load Private Keys ---")

    try:
        # load_pem=True dereferences any values ending in .pem
        keys = load_private_keys(pvtkey_path, load_pem=True)
        print(f"  Loaded {len(keys)} key(s) from: {pvtkey_path}")
        for name in keys:
            value = keys[name]
            preview = (
                value[:40].replace("\n", "\\n") + "..." if len(value) > 40 else value
            )
            print(f"    {name}: {preview}")
        return keys
    except FileNotFoundError:
        print(f"  Key file not found: {pvtkey_path}")
        print(
            "  Create one with: python scripts/key_manager.py add-key <name> <pem-path>"
        )
        return {}


def demo_decrypt_log_entry(keys: dict[str, str]) -> None:
    """Demonstrate decrypt_log_entry()."""
    print("\n--- decrypt_log_entry() ---")
    print("  Decrypts a single log entry dict returned by the API.")
    print("  Returns the entry with decrypted fields merged in (enc field removed).")
    print()
    print("  Signature:")
    print("    decrypt_log_entry(entry, keys, show_enc_block=False)")
    print()
    print("  entry:          dict with 'enc', 'key_name', 'encryption_type' fields")
    print("  keys:           dict of key_name → key_data (from load_private_keys)")
    print("  show_enc_block: if True, keep the raw 'enc' field in the result")
    print()

    if not keys:
        print("  [No keys loaded — showing call pattern only]")
        print("  Example:")
        print("    entry = {")
        print("        'enc': '<base64 ciphertext>',")
        print("        'key_name': 'mykey',")
        print("        'encryption_type': 'Standard',")
        print("        'create_ts': '2026-01-15T10:30:00Z',")
        print("    }")
        print("    result = decrypt_log_entry(entry, keys)")
        print("    # result contains decrypted fields merged into the entry dict")
        return

    # Try to decrypt the first entry that has a matching key
    for entry in SAMPLE_ENC_ENTRIES:
        key_name = entry.get("key_name", "")
        if key_name in keys:
            print(f"  Attempting to decrypt entry with key '{key_name}'...")
            result = decrypt_log_entry(entry, keys)
            if result is not None:
                print(f"  Success. Decrypted fields: {list(result.keys())}")
            else:
                print(
                    "  No result (key matched but decryption failed — check ciphertext)"
                )
            return

    print("  No sample entries matched the loaded keys.")
    print(f"  Loaded keys: {list(keys.keys())}")
    print(
        "  Sample entries use key names: "
        f"{list({e['key_name'] for e in SAMPLE_ENC_ENTRIES})}"
    )


def demo_decrypt_logs(keys: dict[str, str]) -> None:
    """Demonstrate decrypt_logs()."""
    print("\n--- decrypt_logs() ---")
    print("  Decrypts a list of log entries, skipping those with missing keys.")
    print()
    print("  Signature:")
    print("    decrypt_logs(logs, keys, show_enc_block=False)")
    print()
    print("  logs:  list of entry dicts (each with enc, key_name, encryption_type)")
    print("  keys:  dict from load_private_keys()")
    print()

    if not keys:
        print("  [No keys loaded — showing call pattern only]")
        print("  Example:")
        print("    keys = load_private_keys('pvtkey.txt', load_pem=True)")
        print("    decrypted = decrypt_logs(api_log_entries, keys)")
        print(
            "    print(f'Decrypted {len(decrypted)} of {len(api_log_entries)} entries')"
        )
        return

    result = decrypt_logs(SAMPLE_ENC_ENTRIES, keys)
    decrypted = cast(list[dict[str, Any]], result)
    total = len(SAMPLE_ENC_ENTRIES)
    success = len(decrypted)
    skipped = total - success
    print(f"  Input:     {total} entries")
    print(f"  Decrypted: {success} entries")
    print(f"  Skipped:   {skipped} entries (key missing or wrong key)")

    if decrypted:
        print("\n  Sample decrypted entry fields:")
        for key in list(decrypted[0].keys())[:8]:
            print(f"    {key}: {decrypted[0][key]}")


def demo_decrypt_video(
    keys: dict[str, str],
    video_path: str | None,
    output_path: str,
    key_name: str | None,
) -> None:
    """Demonstrate decrypt_video_file()."""
    print("\n--- decrypt_video_file() ---")
    print(
        "  Decrypts a Standard-encrypted video file using chunked I/O (32 MB chunks)."
    )
    print("  Uses atomic write: decrypts to a temp file, then moves it on success.")
    print()
    print("  Signature:")
    print("    decrypt_video_file(video_path, output_path, key_name,")
    print("                       key_file=None, keys=None)")
    print()
    print("  video_path:  path to the encrypted .enc file")
    print("  output_path: destination path for decrypted output")
    print("  key_name:    name of the key in pvtkey.txt to use")
    print("  key_file:    path to pvtkey.txt  (or pass keys= dict directly)")
    print()

    if not video_path:
        print("  [No --video path provided — showing call pattern only]")
        print("  Example:")
        print("    success = decrypt_video_file(")
        print("        'recording.enc',")
        print("        'recording.mp4',")
        print("        key_name='mykey',")
        print("        keys=keys,")
        print("    )")
        print("    if success:")
        print("        print('Video decrypted successfully')")
        return

    if not keys:
        print("  [No keys loaded — cannot decrypt video]")
        return

    if not key_name:
        print("  [No key name provided — use --key-name to specify]")
        return

    print(f"  Video:  {video_path}")
    print(f"  Output: {output_path}")
    print(f"  Key:    {key_name}")
    print()

    try:
        result = decrypt_video_file(
            video_path=video_path,
            output_path=output_path,
            key_name=key_name,
            keys=keys,
        )
        if result:
            print("  Video decrypted successfully.")
    except (FileNotFoundError, KeyError, ValueError, ImportError) as e:
        print(f"  Error: {e}")


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Log Decryption Demo",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
    )
    parser.add_argument(
        "--pvtkey",
        type=str,
        metavar="PATH",
        default=None,
        help="Path to pvtkey.txt (key-name=value format)",
    )
    parser.add_argument(
        "--video",
        type=str,
        metavar="PATH",
        default=None,
        help="Path to an encrypted video file (.enc) to decrypt",
    )
    parser.add_argument(
        "--video-out",
        type=str,
        metavar="PATH",
        default="decrypted_video.mp4",
        help="Output path for decrypted video (default: decrypted_video.mp4)",
    )
    parser.add_argument(
        "--key-name",
        type=str,
        metavar="NAME",
        default=None,
        help="Key name to use for video decryption",
    )
    args = parser.parse_args()

    print("Log Decryption Demo")
    print("=" * 60)

    # --- Key file format ---
    show_key_file_format()

    # --- Load keys ---
    keys = {}
    if args.pvtkey:
        keys = demo_load_keys(args.pvtkey)
    else:
        print("\n--- Load Private Keys ---")
        print("  No --pvtkey provided. Run with:")
        print("    python examples/decrypt_demo.py --pvtkey /path/to/pvtkey.txt")

    # --- decrypt_log_entry ---
    demo_decrypt_log_entry(keys)

    # --- decrypt_logs ---
    demo_decrypt_logs(keys)

    # --- decrypt_video_file ---
    demo_decrypt_video(
        keys=keys,
        video_path=args.video,
        output_path=args.video_out,
        key_name=args.key_name,
    )

    # --- Integration pattern ---
    print("\n--- Full Integration Pattern ---")
    print("  Typical workflow after extracting encrypted logs from the API:")
    print()
    print("    from silo_sdk import LogExtractionAPI, load_config")
    print("    from silo_sdk.logging.decrypt import decrypt_logs, load_private_keys")
    print()
    print("    config = load_config('config/default.json')")
    print("    logs_api = LogExtractionAPI(config)")
    print()
    print("    # 1. Extract encrypted ENC logs")
    print("    raw = logs_api.extract_all_logs(org, start_seq=0, log_types=['ENC'])")
    print()
    print("    # 2. Load private keys")
    print("    keys = load_private_keys('pvtkey.txt', load_pem=True)")
    print()
    print("    # 3. Decrypt in place — merges decrypted fields into each entry")
    print("    decrypted = decrypt_logs(raw, keys)")
    print()
    print("    # 4. Use decrypted logs normally")
    print("    for entry in decrypted:")
    print("        print(entry.get('url'), entry.get('username'))")

    print("\n" + "=" * 60)
    print("Decryption Demo Complete")
    print('  Install extras: pip install "silo-sdk[decrypt]"')
    print("  Manage keys:    python scripts/key_manager.py --help")
    print("  Export logs:    python scripts/export_logs.py --help")


if __name__ == "__main__":
    main()

Usage:

python examples/decrypt_demo.py

Web Harvesting¶

Synchronous Demo¶

HarvesterAPI walkthrough — create task, poll for completion, sync style.

Requires: A8_SCRAPE_TOKEN, A8_FILE_TOKEN, BUCKET_ID

examples/harvest_demo_sync.py

#!/usr/bin/env python3
"""
Sync Harvest Demo.
=================
Demonstrates synchronous harvesting: creating tasks, monitoring
completion, and downloading results.

Usage:
    python examples/harvest_demo_sync.py

Requirements:
    - A8_SCRAPE_TOKEN must be set in config or environment
    - A8_BUCKET_ID must be set in config or environment

Modified: 2025-02-10
"""

import sys
from pathlib import Path

from silo_sdk import HarvesterAPI, load_config
from silo_sdk.base.exceptions import (
    ConfigurationError,
    HarvesterAPIError,
    ValidationError,
)


def main() -> None:
    if "--help" in sys.argv or "-h" in sys.argv:
        print(__doc__)
        sys.exit(0)

    print("Sync Harvest Demo")
    print("=" * 60)

    # Load configuration
    config = load_config("config/default.json")

    # Check required tokens
    if not config.get("SCRAPE_TOKEN"):
        print("Skipping: SCRAPE_TOKEN not configured.")
        print("Set A8_SCRAPE_TOKEN in your .env or config/default.json")
        sys.exit(0)

    if not config.get("BUCKET_ID"):
        print("Skipping: BUCKET_ID not configured.")
        print("Set A8_BUCKET_ID in your .env or config/default.json")
        sys.exit(0)

    try:
        harvester = HarvesterAPI(config)
        print("HarvesterAPI initialized")

        egress_info = config.get("DEFAULT_EGRESS_INFO", {"name": "New York, NY"})
        dest_path = config.get("DEFAULT_DEST_PATH", "/")

        # --- Valid Task Types ---
        print("\n--- Valid Task Types ---")
        task_types = HarvesterAPI.get_valid_task_types()
        print(f"Supported: {sorted(task_types)}")

        # --- Create Asset Harvest Task ---
        print("\n--- Create Asset Harvest Task ---")
        asset_params = {
            "recursive": True,
            "random-wait": True,
            "timeout": 30,
            "level": 2,
        }

        task_id = harvester.create_harvest_task(
            task_type="asset",
            urls=["https://www.example.com"],
            egress_info=egress_info,
            dest_path=dest_path,
            task_params=asset_params,
        )
        print(f"Task created: {task_id}")

        # --- Monitor Task ---
        print("\n--- Monitor Task ---")
        print("Waiting for completion (up to 10 min, checking every 15s)...")

        result = harvester.wait_for_completion(
            task_id=task_id,
            max_wait_time=600,
            check_interval=15,
        )

        if result and result.get("status") == "done":
            print(f"Task completed: {task_id}")
            print(f"  Status: {result['status']}")

            # --- Download Result ---
            if result.get("result_file_id"):
                print("\n--- Download Result ---")
                output_dir = Path("output_files_sync")
                output_dir.mkdir(exist_ok=True)
                output_path = output_dir / f"harvest-{task_id}.zip"

                try:
                    downloaded = harvester.download_task_result(
                        task_id=task_id, output_path=str(output_path)
                    )
                    print(f"Downloaded: {Path(downloaded).name}")
                except Exception as e:
                    print(f"Download failed: {e}")
        else:
            status = result.get("status", "unknown") if result else "no response"
            print(f"Task did not complete successfully. Status: {status}")

        # --- Create Visual Harvest Task ---
        print("\n--- Create Visual Harvest Task ---")
        visual_params = {
            "scale": 1.0,
            "id_egress": True,
            "output_pdf": True,
            "output_image": True,
        }

        vis_task_id = harvester.create_harvest_task(
            task_type="visual",
            urls=["https://www.example.com"],
            egress_info=egress_info,
            dest_path=dest_path,
            task_params=visual_params,
        )
        print(f"Visual task created: {vis_task_id}")

        vis_result = harvester.wait_for_completion(
            task_id=vis_task_id,
            max_wait_time=120,
            check_interval=5,
        )

        if vis_result and vis_result.get("status") == "done":
            print(f"Visual task completed: {vis_task_id}")
        else:
            status = (
                vis_result.get("status", "unknown") if vis_result else "no response"
            )
            print(f"Visual task status: {status}")

        # --- Summary ---
        print("\n" + "=" * 60)
        print("Sync Harvest Demo Summary:")
        print("  - Created and monitored asset harvest task")
        print("  - Created and monitored visual harvest task")
        print("  - Sequential processing demonstrated")
        print("\nFor parallel processing, see harvest_demo_async.py")

    except ConfigurationError as e:
        print(f"Configuration Error: {e}")
        sys.exit(1)
    except ValidationError as e:
        print(f"Validation Error: {e}")
        sys.exit(1)
    except HarvesterAPIError as e:
        print(f"Harvester API Error: {e}")
        sys.exit(1)


if __name__ == "__main__":
    main()

Usage:

python examples/harvest_demo_sync.py

Async Demo¶

Same workflow using async polling — better for bulk task monitoring.

examples/harvest_demo_async.py

#!/usr/bin/env python3
"""
Async Harvest Demo.
==================
Demonstrates asynchronous harvesting: parallel task creation, concurrent
monitoring, and bulk downloads.

Usage:
    python examples/harvest_demo_async.py

Requirements:
    - A8_SCRAPE_TOKEN must be set in config or environment
    - A8_BUCKET_ID must be set in config or environment

Modified: 2026-03-25
"""

import asyncio
import sys
from pathlib import Path

from silo_sdk import HarvesterAPI, load_config
from silo_sdk.base.exceptions import (
    ConfigurationError,
    HarvesterAPIError,
    ValidationError,
)


async def main() -> None:
    if "--help" in sys.argv or "-h" in sys.argv:
        print(__doc__)
        sys.exit(0)

    print("Async Harvest Demo")
    print("=" * 60)

    # Load configuration
    config = load_config("config/default.json")

    # Check required tokens
    if not config.get("SCRAPE_TOKEN"):
        print("Skipping: SCRAPE_TOKEN not configured.")
        print("Set A8_SCRAPE_TOKEN in your .env or config/default.json")
        sys.exit(0)

    if not config.get("BUCKET_ID"):
        print("Skipping: BUCKET_ID not configured.")
        print("Set A8_BUCKET_ID in your .env or config/default.json")
        sys.exit(0)

    try:
        harvester = HarvesterAPI(config)
        print("HarvesterAPI initialized")

        egress_info = config.get("DEFAULT_EGRESS_INFO", {"name": "New York, NY"})
        dest_path = config.get("DEFAULT_DEST_PATH", "/")
        output_dir = "output_files_async"
        Path(output_dir).mkdir(exist_ok=True)

        # --- Single Async Workflow ---
        print("\n--- Single Async Workflow ---")
        asset_params = {
            "recursive": True,
            "random-wait": True,
            "timeout": 30,
            "level": 2,
        }

        # Wrap the workflow call with asyncio.wait_for so that a slow harvest
        # does not block indefinitely.  asyncio.TimeoutError is caught below
        # and treated as a non-fatal warning so the demo continues.
        WORKFLOW_TIMEOUT = 120  # seconds — adjust for real jobs
        try:
            result = await asyncio.wait_for(
                harvester.process_harvest_workflow_async(
                    task_type="asset",
                    urls=["https://www.example.com"],
                    egress_info=egress_info,
                    output_dir=output_dir,
                    dest_path=dest_path,
                    task_params=asset_params,
                    max_wait_time=600,
                    check_interval=15,
                ),
                timeout=WORKFLOW_TIMEOUT,
            )
        except TimeoutError:
            print(
                f"Warning: single workflow timed out after {WORKFLOW_TIMEOUT}s. "
                "The task may still be running in the background."
            )
            result = None

        if result:
            print(f"Task completed: {result['task_id']}")
            print(f"  Status: {result['status']}")
            if result.get("downloaded_file"):
                print(f"  Downloaded: {Path(result['downloaded_file']).name}")
        else:
            print("Asset task did not complete successfully")

        # --- Bulk Async Processing ---
        print("\n--- Bulk Async Processing ---")
        tasks = [
            {
                "task_type": "asset",
                "urls": ["https://www.example.com"],
                "task_params": asset_params,
                "dest_path": dest_path,
                "dest_name": "harvest-asset-<taskid>.zip",
            },
            {
                "task_type": "visual",
                "urls": ["https://www.example.com"],
                "task_params": {
                    "scale": 1.0,
                    "id_egress": True,
                    "output_pdf": True,
                    "output_image": True,
                },
                "dest_path": dest_path,
                "dest_name": "harvest-visual-<taskid>.zip",
            },
        ]

        # bulk_create_and_monitor_async internally uses asyncio.gather with
        # return_exceptions=True, so individual task failures never abort the
        # whole batch.  The call always returns a list of the same length as
        # `tasks`; failed items are None.
        #
        # Progress while tasks run: pass a progress_callback to
        # wait_for_completion_async (used inside process_harvest_workflow_async)
        # to receive a status dict after each polling interval.  Example:
        #
        #   def on_progress(task_info):
        #       print(f"  [{task_info['task_id']}] status={task_info['status']}")
        #
        # Then pass progress_callback=on_progress to process_harvest_workflow_async.
        print(f"Processing {len(tasks)} tasks concurrently...")
        results = await harvester.bulk_create_and_monitor_async(
            tasks=tasks,
            output_dir=output_dir,
            max_concurrent=2,
            egress_info=egress_info,
        )

        # Separate successes from failures for a clear partial-result summary.
        successful = 0
        task_errors = 0
        for i, res in enumerate(results, 1):
            if res is not None and not isinstance(res, Exception):
                print(f"  Task {i}: {res['task_id']} - {res['status']}")
                if res.get("downloaded_file"):
                    print(f"    Downloaded: {Path(res['downloaded_file']).name}")
                successful += 1
            else:
                task_errors += 1
                err_detail = str(res) if isinstance(res, Exception) else "no result"
                print(f"  Task {i}: failed ({err_detail})")

        print(f"\n{successful}/{len(tasks)} tasks completed successfully")
        if task_errors:
            print(
                f"{task_errors} task(s) failed — partial results above. "
                "Other tasks completed normally."
            )

        # --- Async Methods ---
        print("\n--- Available Async Methods ---")
        async_methods = [
            m for m in dir(harvester) if m.endswith("_async") and not m.startswith("_")
        ]
        for method in async_methods:
            print(f"  harvester.{method}()")

        # --- Summary ---
        print("\n" + "=" * 60)
        print("Async Harvest Demo Summary:")
        print("  - Single async workflow demonstrated")
        print("  - Bulk concurrent processing demonstrated")
        print("  - Parallel task creation and monitoring")
        print("\nFor sequential processing, see harvest_demo_sync.py")

    except ConfigurationError as e:
        print(f"Configuration Error: {e}")
        sys.exit(1)
    except ValidationError as e:
        print(f"Validation Error: {e}")
        sys.exit(1)
    except HarvesterAPIError as e:
        print(f"Harvester API Error: {e}")
        sys.exit(1)


if __name__ == "__main__":
    asyncio.run(main())

Usage:

python examples/harvest_demo_async.py

User Management¶

Requires: A8_SYNC_TOKEN, A8_TOP_ORG

python examples/user_demo.py

Organization Hierarchy¶

Requires: A8_ADMIN_TOKEN, A8_TOP_ORG

python examples/org_demo.py

Configuration¶

No token required

python examples/config_demo.py

Operational Scripts¶

Operational scripts for management and automation using the silo-sdk SDK.

These scripts are not included in the wheel distribution. They are meant to be run from the repo root.

`find_user_orgs.py`¶

Search a top-level organization and all descendant sub-orgs to find which organization each user belongs to.

# Use A8_TOP_ORG from config/.env, or override on the command line
python scripts/find_user_orgs.py
python scripts/find_user_orgs.py --top-org my_company
python scripts/find_user_orgs.py --format json
python scripts/find_user_orgs.py --format csv --output results.csv
python scripts/find_user_orgs.py --input /path/to/usernames.txt
python scripts/find_user_orgs.py --help

`move_users.py`¶

Find users across an org tree and move them to a target organization. Runs in dry-run mode by default.

# Use A8_TOP_ORG / A8_TARGET_ORG from config/.env, or override on the command line
python scripts/move_users.py                                         # dry run
python scripts/move_users.py --execute                               # actually move
python scripts/move_users.py --top-org my_company --target-org dest  # override orgs
python scripts/move_users.py --format json                           # JSON plan output
python scripts/move_users.py --help

`manage_sso.py`¶

Full CLI for all Partner SSO API operations. Wraps OrgManagementAPI SSO methods with six subcommands: create, get, update, delete, enable, disable.

# Create a new org with SSO pre-configured
python scripts/manage_sso.py create \
    --org-name MyOrg \
    --vanity-url my-org \
    --idp-name "Okta" \
    --idp-login-url https://my-company.okta.com/app/sso/saml \
    --idp-cert-file ./idp.crt \
    --enable

# Retrieve SSO config
python scripts/manage_sso.py get --org-name MyOrg
python scripts/manage_sso.py get --org-name MyOrg --format json

# Update SSO config (pass only fields to change)
python scripts/manage_sso.py update --org-name MyOrg --idp-login-url https://new.okta.com/sso
python scripts/manage_sso.py update --org-name MyOrg --add-cert ./new_idp.crt
python scripts/manage_sso.py update --org-name MyOrg --idp-cert-file ./cert1.crt --idp-cert-file ./cert2.crt

# Delete org + SSO config (irreversible — requires --yes)
python scripts/manage_sso.py delete --org-name MyOrg --yes

# Enable / disable SSO
python scripts/manage_sso.py enable  --org-name MyOrg
python scripts/manage_sso.py disable --org-name MyOrg

# Import SSO config from SAML IdP metadata XML (create mode — builds a new org)
python scripts/manage_sso.py import-metadata \
    --metadata-file ./idp-metadata.xml \
    --org-name MyOrg \
    --vanity-url my-org \
    --enable

# Import from a live metadata URL (e.g. Azure AD / Okta federation endpoint)
python scripts/manage_sso.py import-metadata \
    --metadata-url https://login.microsoftonline.com/<tenant>/federationmetadata/2007-06/federationmetadata.xml \
    --org-name MyOrg \
    --vanity-url my-org

# Preview what would be parsed without touching the API
python scripts/manage_sso.py import-metadata \
    --metadata-file ./idp-metadata.xml \
    --org-name MyOrg --vanity-url my-org \
    --dry-run

# Update an existing org's SSO certs/URL from refreshed metadata
python scripts/manage_sso.py import-metadata \
    --metadata-file ./idp-metadata.xml \
    --org-name MyOrg \
    --mode update

python scripts/manage_sso.py --help
python scripts/manage_sso.py import-metadata --help

`bulk_create_contexts.py`¶

Read URLs from input_files/urls.txt and create browsing contexts for each. Outputs numbered launch URLs.

python scripts/bulk_create_contexts.py
python scripts/bulk_create_contexts.py --user analyst@company.com
python scripts/bulk_create_contexts.py --research --egress "New York, NY"
python scripts/bulk_create_contexts.py --help

`session_report.py`¶

Generate a session consumption report for an organization using OrgManagementAPI.get_session_report().

export A8_TOP_ORG="my_company"
python scripts/session_report.py
python scripts/session_report.py --start-date 01012025 --end-date 02012025
python scripts/session_report.py --help

`org_usage_report.py`¶

Generate an org-level usage report showing session activity across an organization hierarchy. Displays provisioned users, active users, session counts, and session duration statistics (total, average, longest, shortest) for each org in a tree-style hierarchy.

# Use A8_TOP_ORG from .env, or override on command line
python scripts/org_usage_report.py --org ExampleOrg
python scripts/org_usage_report.py --org ExampleOrg --depth 3
python scripts/org_usage_report.py --org ExampleOrg --start 01-01-2025 --end 04-21-2026
python scripts/org_usage_report.py --org ExampleOrg --csv report.csv
python scripts/org_usage_report.py --org ExampleOrg --active-only
python scripts/org_usage_report.py --help

Requires: A8_ADMIN_TOKEN

`user_usage_report.py`¶

Generate a per-user usage report across an organization hierarchy. Lists users with their session counts and duration statistics. Useful for identifying inactive users, license utilization analysis, and capacity planning.

# Use A8_TOP_ORG from .env, or override on command line
python scripts/user_usage_report.py --org ExampleOrg
python scripts/user_usage_report.py --org ExampleOrg --depth 3
python scripts/user_usage_report.py --org ExampleOrg --start 01-01-2025 --end 04-21-2026
python scripts/user_usage_report.py --org ExampleOrg --csv report.csv
python scripts/user_usage_report.py --org ExampleOrg --active-only --suspended
python scripts/user_usage_report.py --org ExampleOrg --sort total --limit 50
python scripts/user_usage_report.py --help

Requires: A8_ADMIN_TOKEN and A8_SYNC_TOKEN

`file_manager.py`¶

File management CLI with subcommands for listing, searching, downloading, and inspecting files.

python scripts/file_manager.py list [--limit N]
python scripts/file_manager.py search --name "report*"
python scripts/file_manager.py download --file-id ID --output ./report.pdf
python scripts/file_manager.py info --file-id ID

`export_logs.py`¶

Export logs from an organization to JSON or CSV files with automatic pagination, resume support, date-based filtering, and optional encrypted log decryption.

export A8_TOP_ORG="my_company"

# Extract AUTH and SESSION logs to JSON
python scripts/export_logs.py --log-types AUTH,SESSION

# Extract URL logs to CSV, limit to 1000
python scripts/export_logs.py --log-types URL --format csv --max-logs 1000

# Extract logs from the last 7 days
python scripts/export_logs.py --log-types AUTH --days-back 7

# Extract logs from 30 to 7 days ago
python scripts/export_logs.py --log-types AUTH --days-back 30 --end-days-back 7

# Resume from where last extraction left off
python scripts/export_logs.py --log-types AUTH --resume

# Show sequence info and valid log types
python scripts/export_logs.py --info

# Extract encrypted logs with decryption (requires: pip install "silo-sdk[decrypt]")
python scripts/export_logs.py --log-types ALL --encrypted --decrypt \
    --pvtkey /path/to/pvtkey.txt

# Extract encrypted logs via HTTPS proxy
python scripts/export_logs.py --log-types ALL --encrypted --decrypt \
    --pvtkey /path/to/pvtkey.txt --proxy https://proxy.company.com:8080

# Extract encrypted logs and retain raw ciphertext block in output
python scripts/export_logs.py --log-types ENC --encrypted --decrypt \
    --pvtkey /path/to/pvtkey.txt --show-enc

`decrypt_files.py`¶

Decrypt locally stored encrypted log files (ENC type) using native SDK decryption. Use this when you have already extracted ENC logs to disk and want to decrypt them offline, or re-decrypt with a different key.

# Decrypt all ENC*.json files in the current directory
python scripts/decrypt_files.py --pvtkey pvtkey.txt

# Decrypt files in ./out/ and write to ./decrypted/ as CSV
python scripts/decrypt_files.py --pvtkey pvtkey.txt --dir ./out --output-dir ./decrypted --format csv

# Custom file pattern and retain raw enc block in output
python scripts/decrypt_files.py --pvtkey pvtkey.txt --file-mask "*.enc.json" --show-enc

python scripts/decrypt_files.py --help

Requires: pip install "silo-sdk[decrypt]" for Standard decryption.

`scheduled_visual_harvest.py`¶

Schedule a series of visual harvest tasks (screenshot/PDF) for a single URL, one per day, each queued to run at a calculated time using the run_after API parameter. Each daily submission uses a configurable ±jitter offset so captures land within a randomized window around the nominal time.

# 7-day schedule starting now, London egress, default ±15-min jitter
python scripts/scheduled_visual_harvest.py https://example.com --egress London

# Specific start time, 7 days, New York egress, custom output path
python scripts/scheduled_visual_harvest.py https://target.com \
    --egress "New York, NY" --start 2026-03-01T09:00:00Z \
    --days 7 --jitter 15 --dest-path /daily-captures/

# Also request PDF output for each capture
python scripts/scheduled_visual_harvest.py https://example.com \
    --egress "Dallas, TX" --output-pdf

# Preview the computed schedule without submitting any tasks
python scripts/scheduled_visual_harvest.py https://example.com \
    --egress London --dry-run

# 7-day schedule from now, London egress, image output
python scripts/scheduled_visual_harvest.py https://example.com \
    --egress London --output-image

# PDF + A4 landscape, 14 days, specific start time
python scripts/scheduled_visual_harvest.py https://target.com \
    --egress "New York, NY" --start 2026-03-01T09:00:00Z \
    --days 14 --jitter 30 --output-pdf --paper A4 --landscape

# Mobile emulation with translation
python scripts/scheduled_visual_harvest.py https://example.com \
    --egress "Dallas, TX" --emulate "iPhone 15" --is-mobile \
    --translate-target-lang es --output-image

python scripts/scheduled_visual_harvest.py --help

All 27 vis_params are exposed as CLI flags. See --help for the full list organized by group (output types, page rendering, browser identity, PDF options, capture tuning, translation, security).

Requires: A8_SCRAPE_TOKEN, A8_FILE_TOKEN, and BUCKET_ID in config/default.json.

`key_manager.py`¶

Manage private keys and API tokens for log extraction and decryption.

# Add a private key (from file or raw value)
python scripts/key_manager.py add-key mykey /path/to/private.pem
python scripts/key_manager.py add-key legacy_key raw_passphrase

# Remove a key
python scripts/key_manager.py remove-key mykey

# List all stored keys
python scripts/key_manager.py list-keys

# Manage API tokens
python scripts/key_manager.py set-token abc123...
python scripts/key_manager.py show-token

# Clear all key and token files
python scripts/key_manager.py clear

# Use a custom directory for key/token files
python scripts/key_manager.py --path /secure/dir list-keys

`update_collection_examples.py`¶

Update the Postman collection with real API response examples. Runs Newman against the collection, captures live API responses, sanitizes sensitive data (emails, org names, IPs, certs, tokens), and writes the sanitized examples back to the collection JSON.

Maintains a state file (.collection_update_state.json) to reuse the persistent SSO test org across runs — the org is created once and retained until manually deleted.

# Full run — updates all standalone requests, writes collection in place
python scripts/update_collection_examples.py

# Preview what would change without writing the collection
python scripts/update_collection_examples.py --dry-run

# Skip cleanup of test resources (file, context, harvest task) after the run
python scripts/update_collection_examples.py --skip-cleanup

# Increase per-request timeout (default: 30000ms)
python scripts/update_collection_examples.py --timeout 60000

# Reset the persistent SSO test org when needed:
python scripts/manage_sso.py delete --org-name <sso_org_name> --yes
rm .collection_update_state.json   # or edit to remove test_sso_org_name

Requires: A8_ADMIN_TOKEN, A8_SYNC_TOKEN, A8_FILE_TOKEN, A8_LOG_TOKEN, A8_SCRAPE_TOKEN, A8_BUCKET_ID, A8_CATCHALL_USER, A8_TOP_ORG, and Newman installed globally (npm install -g newman).

`audit_org_users.py`¶

Compliance audit — identify inactive, suspended, or incomplete user accounts in an org.

python scripts/audit_org_users.py --org myorg
python scripts/audit_org_users.py --org myorg --inactive-days 60 --csv audit.csv
python scripts/audit_org_users.py --org myorg --include-suborgs

Requires: A8_SYNC_TOKEN (and A8_ADMIN_TOKEN if --include-suborgs)

`batch_user_provisioning.py`¶

Bulk create users from a CSV file with dry-run mode.

CSV columns: username,email,given_name,surname,phone (phone optional)

# Preview — shows what would be created without API calls
python scripts/batch_user_provisioning.py --csv users.csv --org myorg --dry-run

# Execute — creates users and saves results to provisioning_results.json
python scripts/batch_user_provisioning.py --csv users.csv --org myorg --execute

Requires: A8_SYNC_TOKEN

`analyze_log_storage.py`¶

Show available log type breakdown — counts, date ranges, and sequence info.

python scripts/analyze_log_storage.py --org myorg
python scripts/analyze_log_storage.py --org myorg --types AUTH,SESSION,URL
python scripts/analyze_log_storage.py --org myorg --types ALL

Requires: A8_LOG_TOKEN

`compare_org_configs.py`¶

Detect configuration drift across orgs in a hierarchy — proxy policy, user counts.

python scripts/compare_org_configs.py --top-org myorg
python scripts/compare_org_configs.py --top-org myorg --json

Requires: A8_ADMIN_TOKEN

Input Files¶

`input_files/usernames.txt`¶

Input file for find_user_orgs.py and move_users.py. One username per line. Lines starting with # are treated as comments.

`input_files/urls.txt`¶

Input file for bulk_create_contexts.py. One URL per line. Lines starting with # are treated as comments.

Environment Variables¶

Variable	Required By	Description
`A8_ADMIN_TOKEN`	`find_user_orgs` `move_users` `manage_sso` `bulk_create_contexts` `session_report` `org_usage_report` `user_usage_report` `update_collection_examples`	Admin API token
`A8_SYNC_TOKEN`	`find_user_orgs` `move_users` `user_usage_report` `update_collection_examples`	Sync API token (user management)
`A8_FILE_TOKEN`	`file_manager` `scheduled_visual_harvest` `update_collection_examples`	File API token
`A8_BUCKET_ID`	`file_manager` `update_collection_examples`	Default storage bucket
`A8_SCRAPE_TOKEN`	`scheduled_visual_harvest` `update_collection_examples`	Harvest API token
`A8_LOG_TOKEN`	`export_logs` `update_collection_examples` `analyze_log_storage`	Log extraction API token
`A8_TOP_ORG`	`find_user_orgs` `move_users` `manage_sso` `session_report` `export_logs` `update_collection_examples` `audit_org_users` `batch_user_provisioning` `compare_org_configs`	Top-level organization to search
`A8_CATCHALL_USER`	`update_collection_examples`	Default Silo username for browsing contexts
`A8_TARGET_ORG`	`move_users`	Destination organization (also configurable via `--target-org`)
`A8_LOG_LEVEL`	All (optional)	SDK log level (default: INFO)

Examples & Scripts¶

Prerequisites¶

Quick Reference¶

Browsing Isolation¶

File Storage¶

Log Extraction & Decryption¶

Log Extraction Demo¶

Decryption Demo¶

Web Harvesting¶

Synchronous Demo¶

Async Demo¶

User Management¶

Organization Hierarchy¶

Configuration¶

Operational Scripts¶

find_user_orgs.py¶

move_users.py¶

manage_sso.py¶

bulk_create_contexts.py¶

session_report.py¶

org_usage_report.py¶

user_usage_report.py¶

file_manager.py¶

export_logs.py¶

decrypt_files.py¶

scheduled_visual_harvest.py¶

key_manager.py¶

update_collection_examples.py¶

audit_org_users.py¶

batch_user_provisioning.py¶

analyze_log_storage.py¶

compare_org_configs.py¶

Input Files¶

input_files/usernames.txt¶

input_files/urls.txt¶

Environment Variables¶

`find_user_orgs.py`¶

`move_users.py`¶

`manage_sso.py`¶

`bulk_create_contexts.py`¶

`session_report.py`¶

`org_usage_report.py`¶

`user_usage_report.py`¶

`file_manager.py`¶

`export_logs.py`¶

`decrypt_files.py`¶

`scheduled_visual_harvest.py`¶

`key_manager.py`¶

`update_collection_examples.py`¶

`audit_org_users.py`¶

`batch_user_provisioning.py`¶

`analyze_log_storage.py`¶

`compare_org_configs.py`¶

`input_files/usernames.txt`¶

`input_files/urls.txt`¶