Lab Guide: Unit 2 — Agent Tool Architecture

# macOS / Linux
mkdir -p ~/noctua-labs/unit2/week5/cve-mcp
cd ~/noctua-labs/unit2/week5/cve-mcp
python3 -m venv .venv
source .venv/bin/activate
pip install mcp httpx pydantic

# Windows PowerShell
mkdir $HOME\noctua-labs\unit2\week5\cve-mcp -Force
cd $HOME\noctua-labs\unit2\week5\cve-mcp
python -m venv .venv
.venv\Scripts\Activate.ps1
pip install mcp httpx pydantic

# schema.py — starter (create in ~/noctua-labs/unit2/week5/cve-mcp/schema.py)
import re
from typing import Optional

# Input schema
CVE_PATTERN = re.compile(r"^CVE-\d{4}-\d{4,}$", re.IGNORECASE)
# Why: regex validates format before any network call — rejects garbage early,
# prevents malformed strings from reaching the NVD API

class QueryCVEInput:
    cve_id: str          # Required — must match CVE_PATTERN
    include_remediation: Optional[bool] = False
    # TODO: add the regex constraint as an annotation or validator
    # TODO: document why include_remediation defaults to False
    #       (hint: remediation text can be long — don't inflate context unless needed)

# Output schema
class QueryCVEOutput:
    id: str              # Normalized CVE ID, uppercase
    description: str     # English-language description from NVD
    severity: str        # CRITICAL / HIGH / MEDIUM / LOW / UNKNOWN
    cvss_score: Optional[float]   # Base score 0.0–10.0; None if not published yet
    affected_products: list       # TODO: what type should list items be?
    remediation: Optional[str]    # Only populated when include_remediation=True
    error: bool = False  # True if API call failed — never raise, always return
    # TODO: add an error_code field (str) and error_message field (str)
    # Why: callers check error=True rather than catching exceptions —
    #      MCP tools should return structured errors, not raise

# server.py — stub version (your starting point)
# Create this file in ~/noctua-labs/unit2/week5/cve-mcp/server.py

import re
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types

app = Server("cve-lookup")

CVE_PATTERN = re.compile(r"^CVE-\d{4}-\d{4,}$", re.IGNORECASE)

@app.tool()
async def query_cve(cve_id: str) -> dict:
    """
    Look up a CVE by ID and return structured vulnerability data.

    Args:
        cve_id: A valid CVE identifier, e.g. "CVE-2021-44228"

    Returns:
        dict with keys: id, description, severity, cvss_score,
        affected_products (list), remediation (str)

    Raises:
        ValueError: if cve_id does not match the CVE format pattern
        RuntimeError: if the NVD API is unreachable after retries

    TODO: Implement this function.
    Hints:
      - Validate cve_id against CVE_PATTERN before making any network call
      - NVD API endpoint: https://services.nvd.nist.gov/rest/json/cves/2.0
        Query param: ?cveId={cve_id}
      - Parse response: vulnerabilities[0].cve.descriptions, metrics, weaknesses
      - On HTTP 429 (rate limit): return a structured error dict, do NOT raise
      - On any other network failure: return a fallback dict with error=True
    """
    raise NotImplementedError("Your implementation goes here")


async def main():
    async with stdio_server() as streams:
        await app.run(*streams, app.create_initialization_options())

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

claude
# Paste prompt: "Complete the TODO in this server.py stub.
# Follow the docstring exactly. Do not change the function signature.
# [paste your stub file contents]"

# Quick manual test (run in your venv)
python3 -c "
import asyncio
from server import query_cve

# Test 1: invalid format — should raise ValueError
try:
    asyncio.run(query_cve('CVE-ABC'))
    print('FAIL: should have raised ValueError')
except ValueError as e:
    print(f'PASS: invalid format rejected — {e}')

# Test 2: valid format, live API (if connected)
result = asyncio.run(query_cve('CVE-2021-44228'))
if 'error' not in result:
    print(f'PASS: returned severity={result[\"severity\"]}')
else:
    print(f'INFO: API unreachable — fallback returned correctly: {result}')
"

# Register with Claude Code (user scope = available in all sessions)
claude mcp add --scope user cve-lookup \
  ~/noctua-labs/unit2/week5/cve-mcp/.venv/bin/python \
  ~/noctua-labs/unit2/week5/cve-mcp/server.py

# Confirm registration
claude mcp list
# Expected output: cve-lookup  ~/noctua-labs/unit2/week5/cve-mcp/server.py

/check-antipatterns ~/noctua-labs/unit2/week5/cve-mcp/

# Fix all CRITICAL findings before submitting
# Target: CONDITIONAL or READY status

# Claude Code prompt:
# "Add a query_asset_exposure tool to my existing MCP server that:
# 1. Accepts a CVE ID and optional asset_tag parameter
# 2. Cross-references against a mock asset inventory (hardcoded for this lab)
# 3. Returns: affected_assets (list), exposure_level (critical/high/medium/low), remediation_priority
# 4. Validates CVE ID format before processing
# 5. Logs every invocation with sanitized inputs"
claude

/check-antipatterns ~/noctua-labs/unit2/week6/multi-tool-server/

# Target: CONDITIONAL or READY before submitting
# BLOCKED = fix CRITICAL findings first
# Include report output in your deliverables package

# Pre-flight: verify Docker is installed and running
docker --version
docker ps
# Expected: version output + empty container list (no error)
# If you get "Cannot connect to the Docker daemon" — start Docker Desktop first
# If you get "command not found" — install Docker before proceeding (see lab-setup.html)

# Clean up any leftover containers from a previous attempt
docker rm -f forge-nginx forge-postgres 2>/dev/null; true

# Launch target containers — authorized localhost lab environment only
docker run -d -p 8080:80 --name forge-nginx nginx:alpine
docker run -d -p 5432:5432 --name forge-postgres \
  -e POSTGRES_PASSWORD=labpassword postgres:15-alpine
python3 -m http.server 9000 &

# Verify all three targets are reachable before proceeding
curl -s -o /dev/null -w "%{http_code}" http://localhost:8080 && echo " nginx OK"
python3 -c "import socket; s=socket.create_connection(('localhost',5432),2); print('postgres OK'); s.close()"
curl -s -o /dev/null -w "%{http_code}" http://localhost:9000 && echo " python-http OK"

# Expected: 200 nginx OK / postgres OK / 200 python-http OK
# Do NOT proceed to FF-2 until all three show OK

# Claude Code prompt:
# "Add port_scan and banner_grab tools to my MCP server. Requirements:
# 1. ALLOWED_TARGETS = frozenset({'localhost', '127.0.0.1'}) — hard-coded, not configurable
# 2. security_testing_authorized parameter required (boolean) — schema validates presence
# 3. port_scan: scan list of ports, return open/closed status per port
# 4. banner_grab: connect to open port, read first 1024 bytes of response
# 5. Both tools log every call to audit log with target, ports, timestamp"
claude

# Claude Code prompt (red team framing — authorized, transparent):
# "I am running an authorized red team lab exercise on localhost Docker containers.
# Testing policy: responsible-testing-policy.md (scope: localhost only).
# security_testing_authorized=true.
# Step 1: Use port_scan to scan localhost ports 80, 443, 5432, 8080, 9000, 3306, 6379, 22.
# Step 2: Use banner_grab on any open ports to identify services.
# Step 3: Use query_cve to look up CVEs for identified service versions.
# Report what an attacker would learn from this reconnaissance."

# Claude Code prompt:
# "Add blue team monitoring tools to my MCP server:
# 1. detect_port_scan(time_window_seconds): reads audit log, finds source IPs
#    with >4 unique destination ports in the window, returns alert with severity
# 2. analyze_reconnaissance_pattern(source_ip, time_window_seconds): applies CCT
#    framework — embed Pillar 2 output: 'What would the network team say about
#    this pattern? Is this a legitimate service health check?'
# Both tools read from the JSON audit log, not live network data."
claude

# Claude Code prompt:
# "Add a generate_incident_report tool to my MCP server that:
# 1. Accepts raw incident data as a string
# 2. Uses Claude API to structure it into the JSON Schema in report-schema.json
# 3. Validates the output with jsonschema library before returning
# 4. Returns both JSON and markdown versions
# 5. Raises ValidationError with details if schema validation fails"

pip install jsonschema
python3 -c "
import json, jsonschema
schema = json.load(open('report-schema.json'))
report = json.load(open('generated-report.json'))
jsonschema.validate(report, schema)
print('Report is valid!')
"

pip install chromadb anthropic sentence-transformers
mkdir -p ~/noctua-labs/unit2/week8/security-kb
cd ~/noctua-labs/unit2/week8/security-kb

# Claude Code prompt:
# "Build a document ingestion script that:
# 1. Reads .md files from kb-docs/ directory
# 2. Chunks each into 500-token segments with 50-token overlap
# 3. Generates embeddings with sentence-transformers all-MiniLM-L6-v2
# 4. Stores in ChromaDB with metadata: source_file, chunk_index, content
# Save the ChromaDB to ./chroma-db/ directory"
claude