Metadata-Version: 2.4
Name: agent-rbac
Version: 0.2.0
Summary: Role-Based Access Control for AI Agents with ML-Powered Anomaly Detection
Author: AgentRBAC Contributors
License: MIT
Project-URL: Homepage, https://github.com/Parikshit-jadhav-ai/agent-rbac
Project-URL: Documentation, https://github.com/Parikshit-jadhav-ai/agent-rbac#readme
Project-URL: Repository, https://github.com/Parikshit-jadhav-ai/agent-rbac
Project-URL: Issues, https://github.com/Parikshit-jadhav-ai/agent-rbac/issues
Project-URL: Changelog, https://github.com/Parikshit-jadhav-ai/agent-rbac/blob/main/CHANGELOG.md
Keywords: rbac,access-control,ai-agents,llm,security,langchain,crewai,autogen,multi-agent
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Security
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Provides-Extra: ml
Requires-Dist: scikit-learn>=1.3; extra == "ml"
Requires-Dist: numpy>=1.24; extra == "ml"
Requires-Dist: sentence-transformers>=2.2; extra == "ml"
Provides-Extra: langchain
Requires-Dist: langchain>=0.1; extra == "langchain"
Provides-Extra: crewai
Requires-Dist: crewai>=0.1; extra == "crewai"
Provides-Extra: autogen
Requires-Dist: pyautogen>=0.2; extra == "autogen"
Provides-Extra: observability
Requires-Dist: prometheus-client>=0.17; extra == "observability"
Requires-Dist: opentelemetry-sdk>=1.20; extra == "observability"
Requires-Dist: opentelemetry-exporter-otlp>=1.20; extra == "observability"
Provides-Extra: data
Requires-Dist: dask[dataframe]>=2023.1; extra == "data"
Provides-Extra: all
Requires-Dist: agent-rbac[autogen,crewai,data,langchain,ml,observability]; extra == "all"
Provides-Extra: dev
Requires-Dist: pytest>=7; extra == "dev"
Requires-Dist: pytest-cov; extra == "dev"
Dynamic: license-file

<div align="center">

# AgentRBAC

**Role-Based Access Control for AI Agents**

[![Python 3.10+](https://img.shields.io/badge/python-3.10%2B-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-green.svg)](LICENSE)
[![Tests](https://img.shields.io/badge/tests-63%20passed-brightgreen.svg)](#testing)
[![Version](https://img.shields.io/badge/version-0.2.0-orange.svg)](CHANGELOG.md)

A drop-in Python library that gates every agent action — tool calls, memory access, sub-agent spawning — through a layered RBAC policy engine with behavioral ML scoring.

[Installation](#installation) •
[Quick Start](#quick-start) •
[User Guide](#user-guide) •
[API Reference](#api-reference) •
[Examples](examples/)

</div>

---

## What is AgentRBAC?

AgentRBAC provides **fine-grained access control** for AI agent systems. It sits between your agents and their tools/resources, enforcing permissions at every step.

```
Agent request → Behavior tracking → ML risk scoring → Policy evaluation → Capability check → ALLOW / DENY
```

**Key features:**

- **Role hierarchy** with wildcard capabilities and explicit deny rules
- **Registration middleware** (MW1–MW5) for secure agent onboarding
- **Delegation tokens** — scoped, time-bound, HMAC-signed capability delegation
- **ML anomaly detection** — Isolation Forest behavioral scoring with heuristic fallback
- **Multi-agent orchestration** with cumulative risk tracking
- **Append-only audit log** with pluggable backends
- **Framework integrations** — LangChain, CrewAI, AutoGen, or any custom framework
- **Zero required dependencies** — ML and framework extras are opt-in

---

## Installation

### From PyPI (when published)

```bash
pip install agent-rbac
```

### From GitHub

```bash
pip install git+https://github.com/Parikshit-jadhav-ai/agent-rbac.git
```

### From source

```bash
git clone https://github.com/Parikshit-jadhav-ai/agent-rbac.git
cd agent-rbac
pip install -e .
```

### Optional extras

| Extra | Install command | What it adds |
|-------|----------------|--------------|
| **ml** | `pip install "agent-rbac[ml]"` | scikit-learn, sentence-transformers — full ML risk scoring |
| **langchain** | `pip install "agent-rbac[langchain]"` | LangChain tool wrapper |
| **crewai** | `pip install "agent-rbac[crewai]"` | CrewAI task interceptor |
| **autogen** | `pip install "agent-rbac[autogen]"` | AutoGen agent wrapper |
| **observability** | `pip install "agent-rbac[observability]"` | Prometheus + OpenTelemetry |
| **data** | `pip install "agent-rbac[data]"` | Dask for distributed data pipelines |
| **all** | `pip install "agent-rbac[all]"` | Everything above |
| **dev** | `pip install "agent-rbac[dev]"` | pytest + coverage |

---

## Quick Start

```python
import agent_rbac as rbac_lib
from agent_rbac import AgentRBAC, AgentIdentity

# 1. Create the RBAC engine
rbac = AgentRBAC()

# 2. Define roles
rbac.define_role("viewer", capabilities=["tool:*:read"])
rbac.define_role("agent", parent="viewer",
                 capabilities=["memory:*:read", "memory:*:write"])

# 3. Create an agent identity
agent = AgentIdentity(agent_id="my_bot", roles=["agent"])

# 4. Check access
result = rbac.check(agent, "tool:web_search", "read")
print(result.decision)   # Decision.ALLOW
print(result.allowed)    # True
print(result.risk_score) # 0.0
```

### Using the decorator

```python
@rbac.enforce(resource="tool:web_search", action="read")
def web_search(query: str) -> str:
    return f"results for {query}"

result = web_search("AI news", _principal=agent)  # passes RBAC check
```

### Registering agents (v2 middleware)

```python
from agent_rbac import AgentRBAC, AgentType

rbac = AgentRBAC()

agent = rbac.register(
    name="DataProcessor",
    description="Cleans tabular datasets",
    role="data_engineer",
    intent="Process datasets: clean, aggregate, summarise",
    type=AgentType.DATA_AGENT,
    version="1.0.0",
    owner_id="user_123",
    endpoint="http://agents.internal/data-processor",
)

rbac.check(agent, "data", "write").allowed  # True
rbac.check(agent, "model", "train").allowed # False
```

---

## User Guide

### Core Concepts

| Concept | Description |
|---------|-------------|
| **Principal** | Anything that makes a request — a user, an AI agent, or a sub-agent |
| **Role** | Named permission set in a DAG hierarchy. Children inherit parent capabilities |
| **Capability** | Permission string: `resource_type:resource_name:action` with wildcard support |
| **Context** | Runtime state passed to policies: environment, task, timestamps |
| **Decision** | Outcome: `ALLOW`, `DENY`, `DELEGATE`, or `CONDITIONAL` |

### Capability Format

```
tool:web_search:read    # specific tool, specific action
tool:*:read             # all tools, read only
memory:*:*              # all memory, all actions
agent:spawn:*           # spawn any sub-agent type
```

---

### Defining Roles

Roles form a hierarchy. Child roles inherit all parent capabilities. **Explicit deny rules always override allows.**

```python
rbac.define_role("viewer", capabilities=[
    "tool:web_search:read",
    "tool:calculator:read",
])

rbac.define_role("research_agent", parent="viewer",
    capabilities=["tool:arxiv:read", "memory:notes:write"])

rbac.define_role("restricted",
    capabilities=["tool:web_search:read"],
    deny=["tool:payment:*", "memory:credentials:*"])

rbac.define_role("admin",
    capabilities=["tool:*:*", "memory:*:*", "agent:spawn:*"])
```

#### Built-in Role Hierarchy

```
viewer
  ├── developer        (+ agent:create, agent:update, agent:deploy, logs:read)
  ├── data_engineer    (+ data:read, data:write, model:infer)
  └── agent            (+ agent:invoke, data:read, model:infer)
       └── senior_agent (+ data:write, agent:spawn, memory:read)
            └── orchestrator (+ agent:spawn:any, graph:route, pipeline:manage)
                 └── admin   (+ agent:delete, rbac:manage, audit:read, model:train)
```

---

### Checking Access

```python
# Full result object
result = rbac.check(agent, "tool:web_search", "read")
result.decision    # Decision.ALLOW
result.allowed     # True
result.reason      # ""
result.risk_score  # 0.0

# Boolean shorthand
rbac.is_allowed(agent, "tool:web_search", "read")  # True

# Exception-raising variant
from agent_rbac import PermissionDenied
try:
    rbac.check_or_raise(agent, "model", "train")
except PermissionDenied as e:
    print(e.reason)  # "no_matching_capability"
```

---

### Context-Aware Policies

Policies are callables that fire **after** role resolution. Return a `Decision` to override, or `None` to abstain.

```python
from agent_rbac import Decision, Context

def sandbox_guard(principal, resource, action, ctx):
    if ctx.environment == "sandbox" and action == "write":
        return Decision.DENY
    return None

rbac.add_policy("sandbox_guard", sandbox_guard, priority=10)

result = rbac.check(agent, "memory:data", "write",
                    context=Context(environment="sandbox"))
# result.decision == Decision.DENY
```

| Context Field | Type | Description |
|---------------|------|-------------|
| `environment` | `str` | `"production"` / `"staging"` / `"sandbox"` |
| `task_id` | `str` | Current task identifier |
| `task_type` | `str` | Category of the current task |
| `agent_depth` | `int` | Depth in the agent spawn chain |
| `timestamp` | `float` | When the request was made |
| `conversation_hash` | `str` | Hash of conversation history |
| `extra` | `dict` | Additional key-value pairs |

---

### Delegation — Sub-Agent Tokens

A parent agent can delegate a **subset** of its capabilities with time-bound, HMAC-signed tokens.

```python
token = parent.delegate(
    to="summarizer_001",
    capabilities=["tool:web_search:read"],
    roles=["viewer"],
    expires_in=300,  # seconds
)

assert parent.verify_token(token)

child = token.to_identity()
rbac.check(child, "tool:web_search", "read").allowed   # True
rbac.check(child, "memory:data", "write").allowed       # False — exceeds ceiling
```

---

### ML Risk Scoring

AgentRBAC tracks each agent's behavior in a sliding window and scores anomalies from `0` (normal) to `1` (anomalous). Scoring activates after **10 events** per agent.

| Risk Level | Score | Action |
|------------|-------|--------|
| Low | < 0.2 | Pass through |
| Medium | 0.2 – 0.7 | Flagged + logged |
| High | > 0.7 | Overridden to `CONDITIONAL` |

```python
# Adjust thresholds
rbac.set_risk_thresholds(low=0.15, high=0.6)

# Train on custom data
rbac.train_ml_model(samples, contamination=0.05)
rbac.save_ml_model("models/risk_model.pkl")
rbac.load_ml_model("models/risk_model.pkl")
```

**Feature vector** (6 dimensions):

| Feature | Description |
|---------|-------------|
| `action_rate` | Actions per minute in sliding window |
| `unique_resource_ratio` | Distinct resources / total actions |
| `boundary_probe_rate` | Ratio of resource types probed |
| `repeat_ratio` | How often current resource is re-hit |
| `time_since_last` | Seconds since previous action |
| `resource_entropy` | Shannon entropy of resource access distribution |

---

### Multi-Agent Orchestration

```python
orch = rbac.orchestrator(cumulative_risk_ceiling=0.85)
orch.pipeline([gateway.agent_id, processor.agent_id, model.agent_id])

result = orch.run(
    input={"dataset_path": "/data/raw/sales.csv"},
    caller=user_identity,
)

result["hops"]              # per-hop RBAC results
result["cumulative_risk"]   # accumulated risk score
result["short_circuited"]   # True if risk ceiling was hit
result["output"]            # final agent output
```

---

### DataPipeline

```python
from agent_rbac import DataPipeline

pipeline = (
    DataPipeline(source="s3://bucket/raw/sales.csv", partition_size="256MB")
    .clean(drop_nulls=True, deduplicate=True)
    .aggregate(group_by="user_id", agg={"spend": "sum"})
    .write(sink="s3://bucket/processed/", format="parquet")
)

pipeline.plan()                        # inspect steps without running
pipeline.execute(mode="local")         # run locally
pipeline.execute(mode="distributed")   # run with Dask
```

---

### Audit Logging

Every decision is recorded in an append-only audit log with full context.

```python
from agent_rbac import AgentRBAC, FileBackend, StdoutBackend, MemoryBackend

# File backend
rbac = AgentRBAC(audit_file="rbac_audit.jsonl")

# In-memory (testing)
mem = MemoryBackend()
rbac.audit.add_backend(mem)

rbac.check(agent, "tool:web_search", "read")
print(mem.records[-1])
# {'timestamp': ..., 'event': 'access_check', 'agent_id': 'bot_001',
#  'decision': 'ALLOW', 'risk_score': 0.0, ...}

# Custom events
rbac.audit.log_event("agent_spawned", {"child_id": "sub_001"})
```

#### Custom Backend

```python
class MyBackend:
    def write(self, record: dict):
        my_db.insert(record)
    def flush(self):
        pass

rbac.audit.add_backend(MyBackend())
```

---

### Framework Integrations

#### LangChain

```python
from agent_rbac.integrations.langchain import LangChainRBACToolWrapper

wrapped = LangChainRBACToolWrapper(
    rbac=rbac, tool=search_tool,
    resource="tool:duckduckgo", action="read",
)
result = wrapped.run("query", principal=agent, context=ctx)
```

#### CrewAI

```python
from agent_rbac.integrations.crewai import CrewAITaskInterceptor

interceptor = CrewAITaskInterceptor(rbac=rbac)
interceptor.before_task(
    agent_id="researcher",
    task_description="Find papers",
    tools_needed=["tool:arxiv:read"],
    principal=agent, context=ctx,
)
```

#### AutoGen

```python
from agent_rbac.integrations.autogen import AutoGenRBACAgent

guarded = AutoGenRBACAgent(
    rbac=rbac, agent=my_autogen_agent, principal=agent,
    tool_resource_map={"code_exec": "tool:code", "web_search": "tool:web_search"},
)
reply = guarded.generate_reply(messages=[...])
```

---

## Docker

```bash
# Build and run
docker build -t agent-rbac:latest .
docker run --rm agent-rbac:latest

# With persistent audit log
docker run --rm -v $(pwd)/logs:/app/logs \
  -e RBAC_AUDIT_FILE=/app/logs/rbac_audit.jsonl agent-rbac:latest

# docker-compose
docker-compose up                              # start demo
docker-compose run --rm rbac pytest tests/ -v  # run tests
docker-compose --profile ml up                 # with ML deps
```

---

## Testing

```bash
# Install dev dependencies
pip install -e ".[dev]"

# Run tests
pytest tests/ -v

# With coverage
pytest tests/ -v --cov=agent_rbac --cov-report=term-missing

# Run examples
python examples/quickstart.py
python examples/test_agent_demo.py
python examples/delegation.py
python examples/audit_demo.py
```

---

## API Reference

### `AgentRBAC` — Main Engine

| Method | Returns | Description |
|--------|---------|-------------|
| `AgentRBAC(audit_file=None)` | `AgentRBAC` | Create the RBAC engine |
| `.register(**kwargs)` | `AgentIdentity` | Register agent through MW1–MW5 chain |
| `.define_role(name, capabilities, deny, parent)` | `None` | Add a custom role to the hierarchy |
| `.check(principal, resource, action, context)` | `DecisionResult` | Full RBAC + ML pipeline check |
| `.is_allowed(principal, resource, action, ctx)` | `bool` | Boolean shorthand for `.check()` |
| `.check_or_raise(principal, resource, action)` | `DecisionResult` | Raises `PermissionDenied` / `RiskGated` |
| `.add_policy(name, fn, priority)` | `None` | Register a context-aware policy |
| `.bind_role(entity_id, roles, expires_in)` | `None` | Bind roles at runtime with optional TTL |
| `.get_bound_roles(entity_id)` | `list[str]` | Get active (non-expired) bound roles |
| `.orchestrator(cumulative_risk_ceiling)` | `Orchestrator` | Create a pipeline orchestrator |
| `.set_risk_thresholds(low, high)` | `None` | Configure ML gating levels |
| `.train_ml_model(samples, contamination)` | `None` | Train Isolation Forest model |
| `.save_ml_model(path)` | `None` | Persist trained model |
| `.load_ml_model(path)` | `None` | Load trained model |
| `.audit_log(limit)` | `list[dict]` | Retrieve recent audit records |
| `@rbac.enforce(resource, action)` | decorator | Gate function calls with RBAC |

### `AgentIdentity`

| Method | Returns | Description |
|--------|---------|-------------|
| `AgentIdentity(agent_id, roles, capability_ceiling)` | `AgentIdentity` | Create a principal |
| `.delegate(to, capabilities, roles, expires_in)` | `DelegationToken` | Mint a scoped child token |
| `.verify_token(token)` | `bool` | Verify HMAC signature |

### `DelegationToken`

| Method | Returns | Description |
|--------|---------|-------------|
| `.to_identity()` | `AgentIdentity` | Convert token to usable identity |

### `DecisionResult`

| Attribute | Type | Description |
|-----------|------|-------------|
| `decision` | `Decision` | `ALLOW`, `DENY`, `CONDITIONAL`, or `DELEGATE` |
| `allowed` | `bool` | `True` only when decision is `ALLOW` |
| `reason` | `str` | Human-readable explanation |
| `risk_score` | `float` | ML/heuristic risk score (0–1) |

### `Orchestrator`

| Method | Returns | Description |
|--------|---------|-------------|
| `.pipeline(agent_ids)` | `None` | Define a linear agent pipeline |
| `.run(input, caller, context)` | `dict` | Execute with RBAC enforcement at each hop |

### `DataPipeline`

| Method | Returns | Description |
|--------|---------|-------------|
| `DataPipeline(source, partition_size)` | `DataPipeline` | Create a pipeline |
| `.clean(drop_nulls, deduplicate)` | `DataPipeline` | Add cleaning step |
| `.aggregate(group_by, agg)` | `DataPipeline` | Add aggregation step |
| `.filter(expr)` | `DataPipeline` | Add filter step |
| `.write(sink, format)` | `DataPipeline` | Add write step |
| `.plan()` | `list[dict]` | Inspect pipeline steps |
| `.execute(mode)` | `dict` | Run `"local"` or `"distributed"` |

---

## Project Structure

```
agent_rbac/
├── __init__.py              # Public API & exports
├── config.py                # AgentRBAC entry point
├── decorators.py            # @rbac.enforce
├── exceptions.py            # AgentRBACError, PermissionDenied, RiskGated
├── core/
│   ├── identity.py          # AgentIdentity, DelegationToken
│   ├── roles.py             # Role hierarchy (DAG)
│   ├── capabilities.py      # Wildcard capability matching
│   ├── policy.py            # Policy evaluator + Context
│   └── decision.py          # Decision enum + DecisionResult
├── ml/
│   ├── behavior.py          # Sliding window behavior tracker
│   └── risk_scorer.py       # Isolation Forest + heuristic fallback
├── audit/
│   └── logger.py            # FileBackend, StdoutBackend, MemoryBackend
├── registration/
│   └── ...                  # MW1–MW5 middleware chain
├── orchestration/
│   └── orchestrator.py      # Multi-agent pipeline orchestrator
├── data/
│   └── pipeline.py          # DataPipeline fluent API
└── integrations/
    ├── langchain.py         # LangChain tool wrapper
    ├── crewai.py            # CrewAI task interceptor
    └── autogen.py           # AutoGen agent wrapper
tests/
├── test_agent_rbac.py       # Core RBAC tests
└── test_v2_features.py      # v2 feature tests
examples/
├── quickstart.py            # Basic usage
├── test_agent_demo.py       # Full pipeline demo
├── delegation.py            # Delegation tokens
└── audit_demo.py            # Audit logging
```

---

## License

[MIT](LICENSE)

---

<div align="center">
<sub>Built for the age of AI agents. Secure every action.</sub>
</div>
