🛡️ 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

Extra	What it adds	Command
[all]	Everything below combined	pip install "aegisfs[all]"
[crypto]	ECDSA-P256 token signing	pip install "aegisfs[crypto]"
[gateway]	FastAPI REST server + dashboard	pip install "aegisfs[gateway]"
[uuid7]	UUID v7 time-ordered object IDs	pip install "aegisfs[uuid7]"
[dev]	pytest, mypy, ruff linting	pip install "aegisfs[dev]"

---

🚀 Ways to Use AegisFS

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

Access Method	Best For	How to Start
Web Dashboard (FastAPI)	Visual exploration, quick testing	python main.py then open browser
REST API (FastAPI)	Integrations, automated pipelines	python main.py then call endpoints
CLI (Command Line)	Scripting, terminal automation	aegisfs <command>
Python API	Embedding in other applications	from aegisfs.runtime import AegisRuntime

---

🌐 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

URL	What you get
http://localhost:8000/	Interactive Security Control Panel (try all features visually)
http://localhost:8000/docs	Swagger / OpenAPI UI (test every endpoint interactively)
http://localhost:8000/redoc	ReDoc API specification

---

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)

Variable	Default	Description
AEGISFS_JWT_SECRET	aegisfs-default-jwt-secret-...	JWT signing secret (change in production!)
AEGISFS_WORKSPACE	./aegis_gateway_workspace	Workspace directory for the gateway
AEGISFS_ALLOWED_ORIGINS	*	CORS allowed origins (comma-separated)

---

💻 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

Standard	Coverage
draft-aegisfs-secdispatch-rats-01	Full reference implementation
RFC 9334 (RATS Architecture)	Attester, Verifier, Relying Party roles
RFC 2119 / RFC 8174	Requirement levels (MUST, SHOULD, MAY)
PEP 561	py.typed typed package marker

---

📄 License

Distributed under the MIT License.