aegisfs

🛡️ AegisFS (aegisfs)

AI-Driven Programmable Secure File Runtime and Intelligent Workspace Architecture

aegisfs is a reference Python implementation of the AegisFS architecture detailed in IETF Internet-Draft draft-aegisfs-secdispatch-rats-01. Source code and issue tracking are hosted on GitHub at https://github.com/sripad2020/AeGisFS.

It turns passive filesystem objects into intelligent, policy-driven security principals governed by:

• A 25-step execution pipeline for every file operation
• Octal-to-OpCode (OtO) 9-bit cryptographic access control
• IETF RATS EAT device attestation (RFC 9334)
• ECDSA-P256 Capability Tokens with sub-100ms revocation
• Multi-Window AI Digital Twin ransomware detection (1h / 24h / 30d)
• Chain-Hashed Forensics Audit Log (SHA-256 / SHA-3-256 chaining)
• FastAPI Secure Web Gateway with an interactive browser dashboard

---

📦 Installation

Option 1: Install with all features (recommended)

pip install "aegisfs[all]"

Option 2: Install base package only

pip install aegisfs

Option 3: Development install (editable, for modifying the source)

git clone https://github.com/sripad2020/AeGisFS.git
cd AeGisFS
pip install -e ".[all]"

Optional Feature Extras

---

🚀 Ways to Use AegisFS

AegisFS has three access methods — choose whichever fits your workflow:

---

🌐 Method 1: FastAPI Web Gateway (Browser + REST API)

Step 1 — Install gateway dependencies

pip install "aegisfs[gateway]"

Step 2 — Start the server

python main.py

You'll see:

==================================================================
🚀 Starting AegisFS Secure Web Access Gateway (FastAPI Server)
👉 Interactive Dashboard: http://localhost:8000/
📚 OpenAPI Documentation: http://localhost:8000/docs
==================================================================

Step 3 — Open in your browser

---

Interactive Control Panel Features (http://localhost:8000/)

The dashboard gives you point-and-click access to every AegisFS feature:

🔑 JWT Session Authentication

• Select a principal identity (alice_dev, or any name you like)
• Select a group role: developers, admin, or untrusted
• Click "Generate JWT Session Token" — your 15-minute session token appears instantly

🚀 Execute Governed File Operations

• Type any file path (e.g. docs/classified_report.txt)
• Choose the operation: WRITE (atomic staged transaction) or READ (integrity verified)
• Paste your content payload
• Click "Run Pipeline Request" — the full 25-step pipeline executes in real-time

⚡ Live Pipeline Execution Feedback The response panel shows:

• allowed — whether the operation was permitted
• decision — ALLOW, DENY, QUARANTINE, or RESTRICT
• opcode — the 9-bit OtO OpCode e.g. 0o241 (Policy ACL, ACTIVE state, WRITE intent)
• version_id — the hash-linked version number
• content_hash — SHA-256 / SHA-3-256 content integrity hash
• audit_event_id — UUID of the tamper-evident audit event
• risk_score — AI risk score from 0 (safe) to 100 (critical)

---

REST API Endpoints Reference

All endpoints are also accessible as standard REST calls with a JWT Bearer token.

POST /api/v1/auth/token — Issue a JWT Session Token

curl -X POST "http://localhost:8000/api/v1/auth/token" \
     -H "Content-Type: application/json" \
     -d '{"principal_id": "alice", "group": "developers"}'

Response:

{
  "access_token": "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJhbGljZSIsImdycCI6ImRldmVsb3BlcnMifQ...",
  "token_type": "bearer",
  "expires_in": 900,
  "principal_id": "alice",
  "group": "developers"
}

---

POST /api/v1/workspace/file — Write a Governed File

curl -X POST "http://localhost:8000/api/v1/workspace/file" \
     -H "Authorization: Bearer <YOUR_TOKEN>" \
     -H "Content-Type: application/json" \
     -d '{
           "file_path": "reports/q4_summary.txt",
           "content": "Classification: CONFIDENTIAL. Quarterly report data."
         }'

Response:

{
  "allowed": true,
  "decision": "ALLOW",
  "opcode": "0o241",
  "object_id": "018f8c4a-2d3e-7b1a-9c4d-e5f6a7b8c9d0",
  "version_id": 2,
  "content_hash": "a3b4c5d6e7f8a9b0c1d2e3f4a5b6c7d8e9f0a1b2c3d4e5f6a7b8c9d0e1f2a3b4",
  "audit_event_id": "9b8a7c6d-5e4f-3a2b-1c0d-e9f8a7b6c5d4",
  "error_message": "",
  "risk_score": 0
}

---

GET /api/v1/workspace/file — Read a Governed File

curl "http://localhost:8000/api/v1/workspace/file?file_path=reports/q4_summary.txt" \
     -H "Authorization: Bearer <YOUR_TOKEN>"

---

GET /api/v1/workspace/status — Workspace Security Health

curl "http://localhost:8000/api/v1/workspace/status"

Response:

{
  "workspace_root": "/path/to/aegis_gateway_workspace",
  "logged_audit_events": 14,
  "resource_summary": {
    "max_memory_bytes": 2147483648,
    "allocated_memory_bytes": 0,
    "open_handles": 0,
    "max_handles": 500
  },
  "digital_twin_frozen": false,
  "audit_chain_intact": true
}

---

GET /api/v1/audit/verify — Cryptographic Audit Chain Verification

curl "http://localhost:8000/api/v1/audit/verify"

Response:

{
  "chain_integrity": "VERIFIED_INTACT",
  "total_logged_events": 14,
  "last_chain_hash": "7f8a9b0c1d2e3f4a5b6c7d8e9f0a1b2..."
}

---

Environment Variables (Optional)

---

💻 Method 2: Command Line Interface (CLI)

After installation, the aegisfs command is available globally.

Initialize a New Workspace

Creates the .aegis/ directory structure with SQLite metadata store, audit log, and policy cache.

aegisfs init ./my_workspace

Output:

[+] AegisFS workspace initialized at: /path/to/my_workspace
    .aegis directory created with metadata, policies, and audit stores.

---

Compile an APL Policy File

Compiles an Aegis Policy Language (.apl) source file into a cryptographically signed policy bundle.

aegisfs compile ./policy.apl --out .aegis/policies/bundle.json

Example policy.apl:

workspace MyProject {
    classification = confidential
    access {
        developers    = [read, write, lock, snapshot, commit]
        security_team = [read, audit]
        admin         = [read, write, lock, rollback, commit, destroy]
    }
    dataflow {
        permits = [AegisWorkspace, SecureExport]
        denies  = [ExternalDrive, PublicCloud, Email, Clipboard]
    }
    ai {
        anomaly_threshold = 75
        digital_twin      = enabled
    }
}

---

Inspect Workspace Security Status

Shows live workspace health, logged events count, and audit chain integrity.

aegisfs status ./my_workspace

Output:

=== AegisFS Workspace Security Status ===
  Workspace Root:       /path/to/my_workspace
  Logged Audit Events:  7
  Audit Chain Integrity: VERIFIED OK

---

Issue a Capability Token

Issues a short-lived, ECDSA-signed capability token binding a principal to specific OpCodes.

aegisfs capability issue \
  --principal alice_dev \
  --object active/config.json \
  --opcodes 0o241 \
  --ttl 600 \
  --uses 10

Output: A signed JSON token with token_id, issued_at, expires_at, signature, and signing_algo.

---

Execute a Governed Operation

Runs a file write or read operation through the full 25-step secured pipeline.

# Write through the governed pipeline
aegisfs run \
  --workspace ./my_workspace \
  --file docs/architecture.txt \
  --op write \
  --content "AegisFS Architecture Document" \
  --principal alice

# Read with integrity verification
aegisfs run \
  --workspace ./my_workspace \
  --file docs/architecture.txt \
  --op read \
  --principal alice

Output:

=== AegisFS Governed Pipeline Result ===
  Allowed: True
  Decision: ALLOW
  OpCode:   0o241
  Object ID: 018f8c4a-2d3e-7b1a-9c4d-e5f6a7b8c9d0
  Version:  2
  Content Hash: a3b4c5d6e7f8a9b0...

---

Verify Audit Chain Integrity

Cryptographically verifies every chained event hash in the forensics audit log.

aegisfs audit verify ./my_workspace

Output:

[+] Audit Chain Integrity Verification PASSED across 14 events.

---

🐍 Method 3: Python API

Embed AegisFS directly into any Python application.

Basic File Governance

from aegisfs.runtime import AegisRuntime
from aegisfs.oto import OtOIntent

# Initialize a workspace (auto-creates .aegis/ metadata directory)
runtime = AegisRuntime.initialize_workspace("./my_workspace")

# WRITE: Execute a governed write operation through the 25-step pipeline
result = runtime.execute_operation(
    principal_id="alice",
    file_path="docs/report.txt",
    intent=OtOIntent.WRITE,
    data=b"Quarterly Report - CONFIDENTIAL",
    principal_group="developers",
)

print(f"Allowed:      {result.allowed}")       # True
print(f"Decision:     {result.decision}")       # ALLOW
print(f"OpCode:       {result.opcode}")         # 0o241
print(f"Version:      {result.version_id}")     # 2
print(f"Content Hash: {result.content_hash}")   # sha256 hex
print(f"Risk Score:   {result.risk_score}/100") # 0

# Verify audit chain integrity
runtime.audit_logger.verify_chain_integrity()   # True

---

Working with OtO OpCodes

from aegisfs.oto import OtOOpCode, OtODomain, OtOState, OtOIntent

# Build a 9-bit OtO OpCode: Domain=POLICY_ACL(2), State=ACTIVE(4), Intent=WRITE(1)
opcode = OtOOpCode.from_fields(
    domain=OtODomain.POLICY_ACL,
    state=OtOState.ACTIVE,
    intent=OtOIntent.WRITE,
)
print(opcode.octal_str)    # 0o241
print(opcode.domain.name)  # POLICY_ACL
print(opcode.state.name)   # ACTIVE
print(opcode.intent.name)  # WRITE

---

Compile an APL Policy

from aegisfs.apl import APLCompiler

compiler = APLCompiler()
bundle = compiler.compile("""
workspace BankProject {
    classification = restricted
    access {
        developers = [read, write, commit]
        admin      = [read, write, commit, destroy]
    }
}
""")

print(f"Bundle ID:     {bundle.bundle_id}")
print(f"Policy Hash:   {bundle.policy_hash[:16]}...")
print(f"Auth OpCodes:  {len(bundle.authorized_opcodes)}")

---

Issue and Validate Capability Tokens

from aegisfs.capability import CapabilityEngine
from aegisfs.oto import OtOOpCode, OtODomain, OtOState, OtOIntent

engine = CapabilityEngine()

# Issue a signed capability token
opcode = OtOOpCode.from_fields(OtODomain.POLICY_ACL, OtOState.ACTIVE, OtOIntent.WRITE)
token = engine.issue_token(
    principal="alice",
    object_id="docs/classified.txt",
    permitted_opcodes=[opcode.raw],
    ttl_seconds=300,    # 5 minutes
    max_uses=5,
)

print(f"Token ID:   {token.token_id}")
print(f"Signed With:{token.signing_algo}")  # ecdsa-p256 or hmac-sha256
print(f"Expires At: {token.expires_at}")

# Revoke the token
engine.revoke_token(token.token_id)

---

RATS EAT Device Attestation

from aegisfs.rats import RATSAttester, RATSVerifier, TrustLevel

# Device side: generate an attestation token
attester = RATSAttester(device_id="device-001")
verifier = RATSVerifier()

nonce = verifier.issue_nonce()   # Fresh nonce for replay prevention
token  = attester.generate_token(nonce, TrustLevel.TRUSTED)

# Verifier side: validate the EAT token
trust_level = verifier.verify_token(token)
print(f"Device trust level: {trust_level}")  # 2 = TRUSTED

---

DataFlow Violation Detection

from aegisfs.dataflow import DataFlowEngine, DataFlowPolicy
from aegisfs.exceptions import AegisDataFlowViolation

policy = DataFlowPolicy(
    workspace_name="SecureProject",
    permits=["AegisWorkspace"],
    denies=["Dropbox", "Email", "ExternalDrive"],
    classification_level=2,
)
engine = DataFlowEngine(policy)

try:
    engine.evaluate_transfer(
        source_path="reports/classified.txt",
        destination="Dropbox/public_share",
        classification=2,
        principal_id="alice",
    )
except AegisDataFlowViolation as e:
    print(f"BLOCKED: {e}")  # Transfer denied by DataFlow policy

---

Multi-Window AI Anomaly Detection

from aegisfs.ai import DigitalTwinEngine

twin = DigitalTwinEngine()

# Simulate 15 rapid write operations (ransomware pattern)
for i in range(15):
    result = twin.record_write()
    if not result:
        print("🚨 RANSOMWARE DETECTED — Writes FROZEN by Digital Twin!")
        print(f"Anomaly score exceeded threshold across 1h/24h/30d baselines.")
        break

---

🧪 Running Tests

# Run all tests
python -m unittest discover tests -v

# Run v0.2.0 robustness upgrade tests specifically
python -m unittest tests.test_upgrades -v

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

Test coverage includes:

• OtO 9-bit OpCode encoding & dispatch (test_oto.py)
• APL compiler: lexer, parser, AST, and bundle generation (test_apl.py)
• Capability token ECDSA signing, revocation, and expiry (test_capability.py)
• Full 25-step runtime pipeline execution (test_runtime.py)
• RATS EAT attestation, nonce replay prevention (test_upgrades.py)
• DataFlow violation interception (test_upgrades.py)
• Multi-window AI baselines & ransomware detection (test_upgrades.py)
• Thread-safe concurrent audit logging (test_upgrades.py)
• SQLite metadata store upsert & retrieval (test_upgrades.py)
• Adaptive per-object locking strategies (test_upgrades.py)
• Resource value parser: "2GB" → 2147483648 (test_upgrades.py)

---

📁 Project Structure

aefs/
├── aegisfs/                    # Core Package
│   ├── __init__.py             # v0.2.0 Exports & Version
│   ├── oto.py                  # 9-Bit OtO OpCode Encoding & Jump-Table
│   ├── apl/                    # APL Compiler Sub-Package
│   │   ├── lexer.py            # APL Tokenizer / Scanner
│   │   ├── parser.py           # Recursive Descent Parser
│   │   ├── ast_nodes.py        # AST Node Definitions
│   │   └── compiler.py         # PolicyBundle Generator & Cache
│   ├── object.py               # Intelligent File Object (IFO)
│   ├── lifecycle.py            # 8-State Lifecycle FSM
│   ├── identity.py             # Multi-Dimensional Identity & DCAL
│   ├── capability.py           # ECDSA-P256 / HMAC Capability Tokens
│   ├── transaction.py          # Adaptive Locking & Versioning
│   ├── graph.py                # Provenance & Dependency Graph
│   ├── resources.py            # Resource Governance Engine
│   ├── ai.py                   # Multi-Window Digital Twin & Canary
│   ├── audit.py                # Chain-Hashed Forensics Logger
│   ├── rats.py                 # RATS EAT Attestation (RFC 9334)
│   ├── dataflow.py             # DataFlow Policy Interceptor
│   ├── store.py                # SQLite MetadataStore & Graph Persistence
│   ├── exceptions.py           # Centralised Exception Hierarchy
│   ├── runtime.py              # 25-Step Governed Execution Pipeline
│   ├── gateway.py              # FastAPI REST Gateway + Dashboard UI
│   ├── cli.py                  # Terminal CLI (`aegisfs` command)
│   └── py.typed                # PEP 561 Type Stub Marker
├── tests/
│   ├── test_oto.py
│   ├── test_apl.py
│   ├── test_capability.py
│   ├── test_runtime.py
│   └── test_upgrades.py        # v0.2.0 Robustness Tests
├── main.py                     # One-command server launcher
├── run_gateway.py              # Alternative server launcher
├── pyproject.toml              # Package metadata & dependencies
└── README.md                   # This file

---

🏗️ Architecture: The 25-Step Execution Pipeline

Every file operation executed through AegisFS traverses these steps automatically:

 1. User / Application Request
 2. Identity Validation (Principal, MFA, Session)
 3. RATS EAT Device Attestation (RFC 9334)
 4. APL Policy + DCAL Rule Evaluation
 5. Capability Token Validation & Consumption
 6. OtO 9-Bit OpCode Generation & Fault Check
 7. Object Lifecycle State Check (FSM Transition)
 8. Intent Declaration & Canary Decoy Check
 9. Dependency Graph Analysis (Impact Set)
10. Resource Availability Check (Memory, Handles)
11. AI Static File Risk Analysis
12. AI Predictive Risk Decision Engine
13. Transaction BEGIN + Pre-Operation Snapshot
14. File Operation Execution
15. Digital Twin Multi-Window Behavioral Monitoring
16. Adaptive Lock Strategy Resolution
17. DataFlow Destination Interception
18. Transaction Validation
19. Atomic COMMIT + Hash-Linked Version Record
20. Digital Twin Model Update
21. Provenance Graph Edge Write
22. Dependency Impact Set Propagation
23. Audit Event Write + SHA-256 Chain Hash Update
24. Resource Handle Release
25. Final Decision Notification

---

📜 Specification Conformance

---

📄 License

Distributed under the MIT License.