aribot 1.5.0
Aribot Security Platform SDK by Aristiun & Ayurak - Threat modeling, compliance, and cloud security APIs
Requires Python: >=3.8
Keywords: api, cloud-security, compliance, devsecops, security, threat-modeling

# Aribot Python SDK

Official Python SDK for the Aribot Security Platform by Aristiun & Ayurak.

[![PyPI](https://img.shields.io/pypi/v/aribot)](https://pypi.org/project/aribot/)
[![Python](https://img.shields.io/pypi/pyversions/aribot)](https://pypi.org/project/aribot/)
[![License](https://img.shields.io/badge/license-MIT-blue.svg)](LICENSE)

## Installation

```bash
pip install aribot
```

## Quick Start

```python
from aribot import Aribot

client = Aribot(api_key="your_api_key")

# Analyze architecture diagram for threats
result = client.threat_modeling.analyze_diagram("architecture.png")
print(f"Found {result['threat_count']} threats")

# Get detailed threats
threats = client.threat_modeling.get_threats(result['diagram_id'])
for threat in threats:
    print(f"[{threat['severity']}] {threat['title']}")

# Digital Twin - List cloud providers
providers = client.digital_twin.get_providers()
for p in providers:
    print(f"Provider: {p['name']} - {p['display_name']}")

# Economics - Get dashboard
dashboard = client.economics.get_dashboard()
print(f"Total Monthly Cost: ${dashboard['company_summary']['total_monthly']}")

# Red Team - Get methodologies
methodologies = client.redteam.get_methodologies()
for m in methodologies:
    print(f"Methodology: {m['name']}")
```

## Features

- **Threat Modeling** - Upload diagrams, detect components, identify threats
- **Compliance Scanning** - ISO 27001, SOC2, GDPR, HIPAA, PCI-DSS, NIST
- **Cloud Security** - Scan AWS, Azure, GCP for misconfigurations
- **Pipeline Security** - SAST, SCA, secrets detection in CI/CD
- **Digital Twin** - Cloud provider integration, resource discovery, health monitoring
- **Economics** - Cost analysis, ROI calculations, market intelligence
- **Red Team** - Attack simulations, methodologies, threat intelligence

## API Reference

### Threat Modeling

```python
# Upload and analyze a diagram
result = client.threat_modeling.analyze_diagram(
    "architecture.png",
    analysis_depth="comprehensive",  # basic, comprehensive, detailed
    wait=True,                       # wait for analysis to complete
    timeout=300                      # max wait time in seconds
)

# List diagrams
diagrams = client.threat_modeling.list(page=1, limit=25)

# Get diagram details
diagram = client.threat_modeling.get(diagram_id)

# Get threats for a diagram
threats = client.threat_modeling.get_threats(diagram_id)

# Get detected components
components = client.threat_modeling.get_components(diagram_id)

# Run AI-powered analysis
ai_result = client.threat_modeling.analyze_with_ai(
    diagram_id,
    analysis_types=["attack_paths", "data_flow"]
)

# Delete a diagram
client.threat_modeling.delete(diagram_id)

# Get dashboard metrics
dashboard = client.threat_modeling.dashboard(period="month")
```

### Compliance Scanning

```python
# Run compliance scan
result = client.compliance.scan(
    diagram_id,
    standards=["ISO27001", "SOC2", "GDPR"],
    include_recommendations=True
)
print(f"Compliance score: {result['overall_score']}%")

# Get compliance report
report = client.compliance.get_report(diagram_id, format="json")

# List available standards
standards = client.compliance.list_standards()

# Get standard details
iso = client.compliance.get_standard("ISO27001")

# List controls for a standard
controls = client.compliance.list_controls("SOC2", category="access_control")

# Get compliance gaps
gaps = client.compliance.get_gaps(diagram_id, standard_id="ISO27001")

# Create custom standard
custom = client.compliance.add_custom_standard(
    name="Internal Security Policy",
    description="Company security requirements",
    controls=[
        {
            "id": "ISP-001",
            "name": "Data Encryption",
            "description": "All data must be encrypted at rest",
            "severity": "high"
        }
    ]
)

# Get compliance dashboard
dashboard = client.compliance.dashboard(period="quarter")
```

### Cloud Security

```python
# Run cloud security scan
scan = client.cloud.scan(
    project_id="123456789012",
    provider="aws",
    services=["iam", "s3", "ec2"],
    compliance_standards=["CIS-AWS"]
)

# Get scan results
scan = client.cloud.get_scan(scan_id)

# List scans
scans = client.cloud.list_scans(provider="aws", status="completed")

# Get findings
findings = client.cloud.get_findings(
    scan_id,
    severity="critical",
    service="s3"
)

# Connect AWS account
account = client.cloud.connect_account(
    provider="aws",
    credentials={
        "role_arn": "arn:aws:iam::123456789012:role/AribotSecurityRole",
        "external_id": "your-external-id"
    },
    name="Production AWS"
)

# Connect GCP project
account = client.cloud.connect_account(
    provider="gcp",
    credentials={
        "service_account_key": "{ ... }",
        "project_id": "my-project-123"
    }
)

# Connect Azure subscription
account = client.cloud.connect_account(
    provider="azure",
    credentials={
        "tenant_id": "...",
        "client_id": "...",
        "client_secret": "..."
    }
)

# List connected accounts
accounts = client.cloud.list_accounts(provider="aws")

# Get remediation steps
remediation = client.cloud.get_remediation(finding_id)

# Resolve a finding
client.cloud.resolve_finding(
    finding_id,
    resolution="fixed",
    notes="Patched in deployment v1.2.3"
)

# Suppress a finding
client.cloud.suppress_finding(
    finding_id,
    reason="Accepted risk per security review",
    duration_days=90
)

# Get cloud security dashboard
dashboard = client.cloud.dashboard(project_id="123456789012")
```

### Pipeline Security

```python
# Create a project
project = client.pipeline.create_project(
    name="my-api",
    repository_url="https://github.com/org/my-api",
    scan_types=["sast", "sca", "secrets"]
)

# Run security scan
result = client.pipeline.scan(
    project_id,
    commit_sha="abc123def456",
    branch="main",
    scan_types=["sast", "sca", "secrets"],
    fail_on_severity="high",
    wait=True
)

if result['status'] == 'failed':
    print("Security gate failed!")
    for finding in result['blocking_findings']:
        print(f"  [{finding['severity']}] {finding['title']}")

# Get scan details
scan = client.pipeline.get_scan(scan_id)

# Get specific finding types
sast_findings = client.pipeline.get_sast_findings(scan_id)
sca_findings = client.pipeline.get_sca_findings(scan_id)
secrets = client.pipeline.get_secrets_findings(scan_id)

# Configure security gates
client.pipeline.configure_gates(
    project_id,
    gates={
        "fail_on_critical": True,
        "fail_on_high": True,
        "max_high_findings": 5,
        "block_secrets": True,
        "required_scan_types": ["sast", "secrets"]
    }
)

# Set baseline (suppress existing findings)
client.pipeline.add_baseline(project_id, scan_id)

# Suppress a finding
client.pipeline.suppress_finding(
    finding_id,
    reason="False positive - validated manually"
)

# Get pipeline dashboard
dashboard = client.pipeline.dashboard(project_id=project_id)
```

### Digital Twin

```python
# Get cloud providers
providers = client.digital_twin.get_providers()
# Returns: [{'id': '...', 'name': 'aws', 'display_name': 'Amazon Web Services', 'connected': False}, ...]

# Get provider health
health = client.digital_twin.get_health()
# Returns: {'status': 'ok', 'configured': True, 'infrastructure': {...}, 'capabilities': [...]}

# Get analytics
analytics = client.digital_twin.get_analytics()
# Returns: {'configured': True, 'graph_statistics': {...}, 'entities_by_type': {...}}

# Get resources
resources = client.digital_twin.get_resources(limit=50)
# Returns: [{'id': '...', 'name': 'my-bucket', 'provider': 'aws', 'resource_type': 's3'}, ...]

# Sync resources from provider
result = client.digital_twin.sync_resources(provider_id='aws-123')
# Returns: {'status': 'syncing', 'resources_found': 150}

# Discover new resources
discovery = client.digital_twin.discover_resources(provider_id='aws-123')
# Returns: {'status': 'discovered', 'new_resources': 25}
```

### Economics

```python
# Get economics dashboard
dashboard = client.economics.get_dashboard(period='month')
# Returns: {'success': True, 'company_summary': {'total_monthly': 5000, 'total_annual': 60000}, ...}

# Get diagram cost analysis
cost = client.economics.get_diagram_cost_analysis(diagram_id)
# Returns: {'monthly_cost': 1500, 'annual_cost': 18000, 'cost_breakdown': [...]}

# Get component cost
component_cost = client.economics.get_component_cost(component_id)
# Returns: {'component': 'EC2 Instance', 'monthly_cost': 200, 'recommendations': [...]}

# Get economic intelligence (includes Strategic Cost Optimization)
intel = client.economics.get_economic_intelligence()
# Returns: {'status': 'success', 'provider': 'aws', 'pricing': {...},
#           'strategic_optimization': {'current': {...}, 'previous': {...}}}

# Refresh Strategic Cost Optimization recommendations (AI-powered)
# Moves current recommendations to "previous" and generates new ones
refresh = client.economics.refresh_recommendations()
# Returns: {'status': 'success', 'message': 'Recommendations refreshed',
#           'current': {...}, 'previous': {...}}

# Get market intelligence
market = client.economics.get_market_intelligence()
# Returns: {'trends': [...], 'benchmarks': {...}, 'recommendations': [...]}

# Calculate ROI
roi = client.economics.calculate_roi(
    investment=100000,
    risks_addressed=['risk-1', 'risk-2'],
    timeframe_days=365
)
# Returns: {'roi_percentage': 250, 'npv': 150000, 'payback_months': 8}
```

### Red Team

```python
# Get threat modeling methodologies
methodologies = client.redteam.get_methodologies()
# Returns: [{'id': 'stride', 'name': 'STRIDE', 'description': '...'}, ...]

# Get simulations
simulations = client.redteam.get_simulations(limit=10)
# Returns: [{'id': '...', 'name': 'APT29 Simulation', 'status': 'completed'}, ...]

# Get threat intelligence
intel = client.redteam.get_intelligence()
# Returns: {'threats': [...], 'indicators': [...], 'campaigns': [...]}

# Generate attack paths for a diagram
paths = client.redteam.generate_attack_paths(
    diagram_id,
    scope='single',
    include_compliance=True
)
# Returns: {'status': 'success', 'paths': [{'title': '...', 'risk_score': 85, 'steps': [...]}]}

# Get security requirements
requirements = client.redteam.get_security_requirements(diagram_id)
# Returns: [{'id': '...', 'requirement': '...', 'priority': 'high'}, ...]
```

### Severity Assignment (AI-powered)

```python
# Estimate AI cost for severity assignment
estimate = client.compliance.estimate_severity_cost(
    scan_id=scan_id,
    account_id=account_id
)
# Returns: {'estimated_tokens': 15000, 'estimated_cost_usd': 0.45, 'violations_count': 150}

# Assign severity using AI (automatically analyzes all violations)
result = client.compliance.assign_severity_ai(
    scan_id=scan_id,
    account_id=account_id,
    model='claude-3-sonnet',  # or 'gpt-4', 'gemini-pro'
    batch_size=50
)
# Returns: {'status': 'completed', 'processed': 150, 'updated': 142, 'errors': 0}

# Manually assign severity to violations
result = client.compliance.assign_severity_manual(
    violation_ids=['v-123', 'v-456'],
    severity='high'  # critical, high, medium, low, info
)
# Returns: {'updated': 2, 'violations': [...]}
```

### Scanner Rules

```python
# List scanner rules
rules = client.compliance.list_scanner_rules(
    severity='critical',
    provider='aws'
)
# Returns: [{'id': '...', 'name': 'S3 Public Access', 'severity': 'critical'}, ...]

# Get scanner rule statistics
stats = client.compliance.get_scanner_statistics()
# Returns: {'total_rules': 500, 'by_severity': {...}, 'by_provider': {...}}

# Sync rules from cloud providers
sync_result = client.compliance.sync_scanner_rules(
    providers=['aws', 'azure', 'gcp']
)
# Returns: {'synced': 150, 'new': 25, 'updated': 10}

# Create custom scanner rule
rule = client.compliance.create_scanner_rule(
    name='Custom S3 Encryption Check',
    description='Ensure all S3 buckets have encryption enabled',
    severity='high',
    provider='aws',
    resource_type='s3_bucket',
    condition={
        'field': 'encryption.enabled',
        'operator': 'equals',
        'value': True
    }
)
```

### Dynamic Cloud Scanning

```python
# Execute dynamic scan on cloud account
scan = client.compliance.execute_dynamic_scan(
    account_id=account_id,
    scan_type='full',  # full, quick, targeted
    resources=['ec2', 's3', 'iam'],
    standards=['CIS-AWS', 'SOC2']
)
# Returns: {'scan_id': '...', 'status': 'running', 'estimated_duration': 300}

# Execute unified scan with flexible scope
scan = client.compliance.execute_unified_scan(
    scope='account',  # account, standard, diagram
    scope_id=account_id,
    include_remediation=True
)
# Returns: {'scan_id': '...', 'status': 'queued'}
```

### Remediation Execution

```python
# Preview remediation before execution
preview = client.compliance.preview_remediation(
    policy_id=policy_id,
    account_id=account_id
)
# Returns: {'actions': [...], 'risk_level': 'low', 'affected_resources': 5}

# Execute remediation
result = client.compliance.execute_remediation(
    policy_id=policy_id,
    account_id=account_id,
    dry_run=False,
    auto_approve=True
)
# Returns: {'status': 'completed', 'resources_fixed': 5, 'rollback_available': True}
```

### AI Agents & Self-Healing

```python
# Get AI agent status
status = client.ai_agents.get_status()
# Returns: {'active': True, 'agents': [...], 'capabilities': [...]}

# Run specialist agent analysis
analysis = client.ai_agents.run_specialist(
    agent_type='security',  # security, compliance, cost, architecture
    diagram_id=diagram_id
)

# Self-healing operations
healing_status = client.ai_agents.get_self_healing_status()
# Returns: {'enabled': True, 'recent_actions': [...], 'autonomy_level': 'supervised'}

# Get autonomy stats
stats = client.ai_agents.get_autonomy_stats()
# Returns: {'total_remediations': 150, 'auto_approved': 120, 'manual_review': 30}

# Trigger remediation
remediation = client.ai_agents.trigger_remediation(
    finding_id=finding_id,
    auto_approve=False
)

# Approve/reject remediation action
client.ai_agents.approve_remediation(remediation_id)
client.ai_agents.reject_remediation(remediation_id, reason='Risk too high')

# Rollback remediation
client.ai_agents.rollback_remediation(remediation_id)

# Emergency stop all autonomous actions
client.ai_agents.emergency_stop()

# Resume autonomous operations
client.ai_agents.resume_operations()
```

### Security Co-Pilot

```python
# Get security co-pilot status
status = client.security_copilot.get_status()
# Returns: {'enabled': True, 'mode': 'supervised', 'active_threats': 5}

# Get pending actions awaiting approval
actions = client.security_copilot.get_pending_actions()
# Returns: [{'id': '...', 'type': 'patch', 'resource': '...', 'risk': 'low'}, ...]

# Get action history
history = client.security_copilot.get_action_history(limit=50)

# Get active threats
threats = client.security_copilot.get_active_threats()

# Approve/reject action
client.security_copilot.approve_action(action_id)
client.security_copilot.reject_action(action_id)

# Rollback action
client.security_copilot.rollback_action(action_id)

# Trigger security scan
scan = client.security_copilot.trigger_scan(scope='full')

# Update settings
client.security_copilot.update_settings(
    auto_remediation=True,
    max_risk_level='medium'
)

# Toggle co-pilot on/off
client.security_copilot.toggle(enabled=True)
```

### Presets Import

```python
# Import compliance presets
presets = client.compliance.import_presets(
    provider='aws',  # aws, azure, gcp, all
    categories=['security', 'cost', 'operations']
)
# Returns: {'imported': 50, 'standards': [...], 'rules': [...]}
```

### Strategic Remediation Plan (AI-powered)

```python
# Generate strategic remediation plan for a severity level
# Uses AI to analyze all violations and create a comprehensive plan
plan = client.compliance.generate_strategic_plan(
    scan_id=scan_id,
    account_id=account_id,
    severity='critical'  # critical, high, medium, low
)
# Returns: {
#   'id': 'uuid',
#   'severity': 'critical',
#   'status': 'generated',
#   'overview': 'Strategic overview of all critical violations...',
#   'root_causes': [{'cause': '...', 'theme': '...', 'impact': '...'}],
#   'grouped_violations': [{'policy': '...', 'violations': [...]}],
#   'high_impact_actions': [{'action': '...', 'impact_score': 95}],
#   'remediation_phases': [{'phase': 1, 'duration': '1 week', 'actions': [...]}],
#   'success_metrics': [{'metric': '...', 'target': '...'}],
#   'estimated_total_effort': '2-3 weeks',
#   'cost_savings_finops': {'monthly': 5000, 'annual': 60000},
#   'cost_savings_risk': {'risk_reduction': '85%', 'avoided_incidents': 12}
# }

# Get existing strategic plan
existing = client.compliance.get_strategic_plan(plan_id)

# List strategic plans for an account
plans = client.compliance.list_strategic_plans(
    account_id=account_id,
    severity='critical'
)
```

### Mitigation Plan (AI-powered)

```python
# Get existing mitigation plan for a diagram
plan = client.threat_modeling.get_mitigation_plan(diagram_id)
if plan:
    print(f"Overview: {plan['plan']['overview']}")
    print(f"Recommendations: {len(plan['plan']['recommendations'])}")
    for rec in plan['plan']['recommendations']:
        print(f"  [{rec['rank']}] {rec['title']} - {rec['priority']}")

# Generate new mitigation plan (AI-powered, uses chunked processing)
plan = client.threat_modeling.generate_mitigation_plan(
    diagram_id,
    force_regenerate=True  # Force regenerate even if cached
)
# Returns comprehensive plan with:
# - overview: Strategic summary of all threats
# - recommendations: Ranked list with code snippets
# - root_causes: Identified patterns across threats
# - remediation_phases: Phased approach with timelines
# - success_metrics: Measurable criteria
# - metadata: AI provider, generation time, etc.

print(f"Generated in {plan['plan']['metadata']['generation_time_ms']}ms")
print(f"Threats analyzed: {plan['plan']['metadata']['threat_count']}")

# Access recommendations with code snippets
for rec in plan['plan']['recommendations']:
    print(f"[{rec['priority']}] {rec['title']}")
    print(f"  Impact: {rec['impact']}")
    print(f"  Effort: {rec['effort']}")
    print(f"  Affected threats: {', '.join(rec.get('affected_threats', []))}")
    if rec.get('code_snippet'):
        print(f"  Code ({rec['code_language']}):")
        print(f"    {rec['code_snippet']}")

# View root causes (patterns identified across multiple threats)
for cause in plan['plan']['root_causes']:
    print(f"Root cause: {cause['cause']}")
    print(f"  Theme: {cause['theme']}, Impact: {cause['impact']}")

# View phased remediation plan
for phase in plan['plan']['remediation_phases']:
    print(f"Phase: {phase['phase']} ({phase['duration']})")
    print(f"  Actions: {', '.join(phase['actions'])}")
    print(f"  Threats resolved: {phase['threats_resolved']}")

# Update plan with manual edits
updated_plan = plan['plan'].copy()
updated_plan['recommendations'][0]['priority'] = 'immediate'
client.threat_modeling.update_mitigation_plan(diagram_id, updated_plan)
```

## Error Handling

```python
from aribot import (
    Aribot,
    AribotError,
    AuthenticationError,
    RateLimitError,
    ValidationError,
    NotFoundError,
    ServerError
)

client = Aribot(api_key="your_api_key")

try:
    result = client.threat_modeling.analyze_diagram("diagram.png")
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")
except ValidationError as e:
    print(f"Invalid request: {e.errors}")
except NotFoundError:
    print("Resource not found")
except ServerError:
    print("Server error - try again later")
except AribotError as e:
    print(f"API error: {e.message}")
```

## Configuration

```python
# Custom base URL (for on-premise deployments)
client = Aribot(
    api_key="your_api_key",
    base_url="https://aribot.internal.company.com/api",
    timeout=60
)

# Check API health
health = client.health()

# Get current user info
user = client.me()

# Get usage stats
usage = client.usage(period="month")
print(f"API calls used: {usage['calls_used']}/{usage['calls_limit']}")
```

## CI/CD Integration

### GitHub Actions

```yaml
- name: Security Scan
  env:
    AYURAK_API_KEY: ${{ secrets.AYURAK_API_KEY }}
  run: |
    pip install aribot
    python -c "
    from aribot import Aribot
    client = Aribot(api_key='$AYURAK_API_KEY')
    result = client.pipeline.scan(
        project_id='${{ vars.PROJECT_ID }}',
        commit_sha='${{ github.sha }}',
        fail_on_severity='high',
        wait=True
    )
    if result['status'] == 'failed':
        exit(1)
    "
```

### GitLab CI

```yaml
security_scan:
  script:
    - pip install aribot
    - python scripts/security_scan.py
  variables:
    AYURAK_API_KEY: $AYURAK_API_KEY
```

## Support

- Documentation: https://developers.aribot.com/docs/python-sdk
- API Reference: https://developers.aribot.com/api
- Issues: https://github.com/AribotAI/aribot-python/issues

## Changelog

### v1.5.0
- Added Strategic Remediation Plan API (`client.compliance.generate_strategic_plan`, `get_strategic_plan`, `list_strategic_plans`) - AI-powered strategic planning for compliance violations with cost savings analysis
- Added Economic Intelligence Refresh (`client.economics.refresh_recommendations`) - Refresh AI-powered cost optimization recommendations
- Added Severity Assignment API (`client.compliance.estimate_severity_cost`, `assign_severity_ai`, `assign_severity_manual`)
- Added Scanner Rules API (`client.compliance.list_scanner_rules`, `get_scanner_statistics`, `sync_scanner_rules`, `create_scanner_rule`)
- Added Dynamic Cloud Scanning (`client.compliance.execute_dynamic_scan`, `execute_unified_scan`)
- Added Remediation Execution (`client.compliance.preview_remediation`, `execute_remediation`)
- Added AI Agents & Self-Healing module (`client.ai_agents`) with autonomous remediation
- Added Security Co-Pilot module (`client.security_copilot`) for supervised security operations
- Added Presets Import (`client.compliance.import_presets`)
- Added Mitigation Plan API (`client.threat_modeling.get_mitigation_plan`, `generate_mitigation_plan`, `update_mitigation_plan`) - AI-powered strategic remediation planning with chunked processing for large threat sets

### v1.4.0
- Updated base URL to `api.aribot.ayurak.com`
- Added AI module (`client.ai`) - usage, quota, models, configure, analyze, queue status
- Added SBOM module (`client.sbom`) - document management and vulnerability scanning
- Added Dashboard module (`client.dashboard`) - overview, recent activity, risk summary
- Added FinOps module (`client.finops`) - cost optimization recommendations and tracking
- Added Marketplace module (`client.marketplace`) - templates, categories, featured content
- Added API Keys module (`client.api_keys`) - key listing and revocation

### v1.1.0
- Added Digital Twin, Economics, and Red Team modules
- Added Remediation module

### v1.0.0
- Initial release with Threat Modeling, Compliance, Cloud Security, and Pipeline modules

## License

MIT