initial deploy — jairaj files, teammate files pending

Dockerfile

@@ -0,0 +1,6 @@
+FROM python:3.10-slim
+WORKDIR /app
+COPY . .
+RUN pip install openenv flask fastapi uvicorn httpx gradio requests bandit
+EXPOSE 7860
+CMD ["openenv", "run", "--port", "7860"]

README.md

@@ -1,10 +0,0 @@
----
-title: Cannon And Wall
-emoji: 😻
-colorFrom: green
-colorTo: blue
-sdk: docker
-pinned: false
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

agents/cannon_prompt.py

@@ -0,0 +1,8 @@
+def build_cannon_prompt(source_code: str, stage: int, round: int) -> str:
+    return f"""You are a Red Team security analyst. Stage {stage}, Round {round}.
+Analyze this source code and find ONE vulnerability.
+Respond ONLY in this exact JSON format:
+{{"agent":"cannon","vuln_type":"sqli|xss|broken_auth","line_number":0,"explanation":"...","proof_of_concept":"..."}}
+
+SOURCE CODE:
+{source_code}"""

agents/wall_prompt.py

@@ -0,0 +1,8 @@
+def build_wall_prompt(source_code: str, cannon_report: dict, stage: int) -> str:
+    return f"""You are a Blue Team security engineer. Stage {stage}.
+Red team found: {cannon_report}
+Patch the vulnerability. Respond ONLY in this exact JSON format:
+{{"agent":"wall","patched_code":"...full fixed file...","explanation":"..."}}
+
+SOURCE CODE:
+{source_code}"""

client/__init__.py

@@ -0,0 +1 @@
+# client package

client/client.py

@@ -0,0 +1,17 @@
+import httpx
+
+class CannonWallClient:
+    def __init__(self, base_url: str = "http://localhost:7860"):
+        self.base_url = base_url
+
+    def reset(self, stage: int = 1) -> dict:
+        r = httpx.post(f"{self.base_url}/reset", params={"stage": stage})
+        return r.json()
+
+    def step(self, action: dict) -> dict:
+        r = httpx.post(f"{self.base_url}/step", json=action)
+        return r.json()
+
+    def state(self) -> dict:
+        r = httpx.get(f"{self.base_url}/state")
+        return r.json()

environment/curriculum.py

@@ -0,0 +1,18 @@
+from pathlib import Path
+
+STAGES = {
+    1: {"file": "environment/vulnerable_app/stage_1/app.py",
+        "vulns": ["sqli"], "description": "Single SQLi in login form"},
+    2: {"file": "environment/vulnerable_app/stage_1/app.py",
+        "vulns": ["xss"], "description": "Single XSS in comment box"},
+    3: {"file": "environment/vulnerable_app/stage_1/app.py",
+        "vulns": ["sqli", "xss", "broken_auth"], "description": "All three vulns"}
+}
+
+def next_stage(current: int, scores: dict) -> int:
+    if scores.get("wall", 0) > 8:
+        return min(current + 1, max(STAGES.keys()))
+    return current
+
+def get_stage_source(stage: int) -> str:
+    return Path(STAGES[stage]["file"]).read_text()

environment/judge/__init__.py

@@ -0,0 +1 @@
+# judge package

environment/server.py

@@ -0,0 +1,45 @@
+from openenv import Environment
+from pathlib import Path
+from environment.models import Observation
+from environment.judge.verifier import verify_cannon, verify_wall
+from environment.judge.reward import cannon_reward, wall_reward
+
+STAGE_FILES = {
+    1: "environment/vulnerable_app/stage_1/app.py"
+}
+GROUND_TRUTH = {
+    1: {"vuln_type": "sqli", "line_number_range": [10, 20]}
+}
+
+class CannonWallEnvironment(Environment):
+    def reset(self, stage: int = 1):
+        source = Path(STAGE_FILES[stage]).read_text()
+        self.state = {
+            "round": 1,
+            "stage": stage,
+            "source_code": source,
+            "scores": {"cannon": 0.0, "wall": 0.0},
+            "last_cannon_report": None
+        }
+        return self.state
+
+    def step(self, action: dict):
+        agent = action.get("agent")
+        if agent == "cannon":
+            v = verify_cannon(action)
+            if not v["valid"]:
+                return {"reward": 0, "issues": v["issues"]}
+            r = cannon_reward(action, GROUND_TRUTH[self.state["stage"]])
+            self.state["scores"]["cannon"] += r
+            self.state["last_cannon_report"] = action
+            return {"reward": r, "valid": True}
+        elif agent == "wall":
+            test_results = verify_wall(
+                action.get("patched_code", ""),
+                self.state["stage"]
+            )
+            r = wall_reward(action.get("patched_code",""), [], test_results)
+            self.state["scores"]["wall"] += r
+            self.state["round"] += 1
+            return {"reward": r, "test_results": test_results}
+        return {"error": "unknown agent"}

environment/vulnerable_app/stage_1/app.py

@@ -0,0 +1,55 @@
+# stage_1/app.py
+# Cannon & Wall -- Vulnerable Target App (Stage 1)
+# WARNING: This file is intentionally vulnerable for security training purposes.
+# DO NOT deploy this in production.
+
+from flask import Flask, request, session, jsonify
+
+app = Flask(__name__)
+app.secret_key = "hardcoded_secret_123"
+
+# Lines 10-20: SQLi vulnerability lives here
+@app.route("/login", methods=["POST"])
+def login():
+    username = request.form.get("username", "")
+    password = request.form.get("password", "")
+    # VULN: sqli -- string interpolation directly into SQL query
+    query = f"SELECT * FROM users WHERE username='{username}' AND password='{password}'"
+    # Attacker input: ' OR 1=1-- bypasses this check entirely
+    users = {"admin": "password123", "user1": "secret"}
+    if username in users and users[username] == password:
+        session["user"] = username
+        return jsonify({"status": "ok", "user": username})
+    return jsonify({"status": "fail"}), 401
+
+# Lines 25-35: XSS vulnerability lives here
+@app.route("/comments", methods=["GET"])
+def comments():
+    comment = request.args.get("comment", "")
+    # VULN: xss -- user input rendered directly into HTML without escaping
+    return f"<html><body><div>{comment}</div></body></html>"
+    # Attacker input: <script>alert('xss')</script> executes in browser
+
+
+
+# blank line above (line 35)
+# blank line above (line 36)
+# blank line above (line 37)
+# blank line above (line 38)
+# Lines 40-55: Broken Auth vulnerability lives here
+@app.route("/dashboard", methods=["GET"])
+def dashboard():
+    # VULN: broken_auth -- session user is set from request without server validation
+    if request.args.get("user"):
+        session["user"] = request.args.get("user")
+    user = session.get("user")
+    if not user:
+        return jsonify({"status": "unauthorized"}), 401
+    return jsonify({"status": "ok", "dashboard": f"Welcome {user}"})
+
+@app.route("/health")
+def health():
+    return jsonify({"status": "running"})
+
+if __name__ == "__main__":
+    app.run(debug=False, port=5001)

openenv.py

@@ -0,0 +1,22 @@
+"""
+Local shim for the OpenEnv Environment base class.
+The PyPI `openenv` package (v0.1.x) exposes only openenv.env.Env
+and does not include an `Environment` class. This module provides
+the abstract base that CannonWallEnvironment subclasses, matching
+the interface declared in openenv.yaml.
+"""
+
+class Environment:
+    """Base class for OpenEnv-compatible environments."""
+
+    def __init__(self):
+        self.state = {}
+
+    def reset(self, **kwargs) -> dict:
+        raise NotImplementedError
+
+    def step(self, action: dict) -> dict:
+        raise NotImplementedError
+
+    def get_state(self) -> dict:
+        return self.state

openenv.yaml

@@ -0,0 +1,13 @@
+name: cannon-and-wall
+version: 1.0.0
+author: Jairaj S
+theme: self-improvement+multi-agent
+base_class: Environment
+entry: environment/server.py
+client: client/client.py
+port: 7860
+tags:
+  - security
+  - red-team
+  - self-play
+  - rl

ui/demo.py

@@ -0,0 +1,29 @@
+import gradio as gr
+from client.client import CannonWallClient
+
+client = CannonWallClient()
+
+def start_round(stage):
+    obs = client.reset(stage=int(stage))
+    return obs["source_code"], "Round started. Source loaded.", ""
+
+def get_scores():
+    s = client.state()
+    scores = s.get("scores", {})
+    return f"Cannon: {scores.get('cannon',0)} | Wall: {scores.get('wall',0)} | Round: {s.get('round',1)}"
+
+with gr.Blocks() as demo:
+    with gr.Tab("Run Episode"):
+        stage_dd = gr.Dropdown([1,2,3], value=1, label="Stage")
+        start_btn = gr.Button("Start Round")
+        source_box = gr.Textbox(label="Source Code", lines=20)
+        status_box = gr.Textbox(label="Status")
+        output_box = gr.Textbox(label="Agent Output")
+        start_btn.click(start_round, inputs=[stage_dd], outputs=[source_box, status_box, output_box])
+    with gr.Tab("Scores"):
+        score_btn = gr.Button("Refresh Scores")
+        score_box = gr.Textbox(label="Current Scores")
+        score_btn.click(get_scores, outputs=[score_box])
+
+if __name__ == "__main__":
+    demo.launch(server_port=7861)