.gitattributes
0 → 100644
+5
−0
team-project-csrf/README.md
0 → 100644
+26
−0
+11
−0
| Original line number | Diff line number | Diff line |
|---|---|---|
client: replay.sh
|
||
----------------------------------------------------------------------------------------
|
||
C0 control: genuine form, same-origin, token present -> wiener.new@ -> HTTP 200
|
||
A1 §3.1 GET /change-email?email=attacker, token dropped, cross-site -> HTTP 405
|
||
A2 §3.2B GET …&_method=POST (method override), cross-site -> HTTP 405
|
||
A3 §3.2A POST, token absent, cross-site form -> HTTP 403
|
||
A4 POST from evil.example.com (same-site sibling), token absent -> HTTP 403
|
||
A5 POST with allowed Origin, token absent (layer 3 alone must hold) -> HTTP 403
|
||
A6 POST with allowed Origin, token from another session -> HTTP 403
|
||
----------------------------------------------------------------------------------------
|
||
final state: current email: [email protected] |
+87
−0
| Original line number | Diff line number | Diff line |
|---|---|---|
# csrf_gate.py — Experiment C: one gate, every unsafe method, nothing routes around it
|
||
# Run: python3 csrf_gate.py (HTTPS on 127.0.0.1:8443, self-signed cert.pem/key.pem)
|
||
# Then: bash replay.sh (replays the attack shapes of §3.1 / §3.2 and a control)
|
||
import hmac, logging, secrets, sys
|
||
from flask import Flask, abort, request, session, render_template_string
|
||
app = Flask(__name__)
|
||
app.secret_key = secrets.token_bytes(32) # per-process for the lab; persist in production
|
||
app.config.update(
|
||
SESSION_COOKIE_SAMESITE="Lax", # layer 1 — enforced by the browser (experiment B)
|
||
SESSION_COOKIE_SECURE=True,
|
||
SESSION_COOKIE_HTTPONLY=True,
|
||
)
|
||
SAFE = {"GET", "HEAD", "OPTIONS"}
|
||
ALLOWED_ORIGINS = {"https://app.example.com"} # exact origins, never bare domains
|
||
logging.basicConfig(stream=sys.stdout, level=logging.INFO, format="%(message)s")
|
||
log = logging.getLogger("gate")
|
||
def reject(status, layer, reason):
|
||
log.info(f"[GATE] REJECT {status} {request.method:<4} {request.full_path.rstrip('?'):<48} layer={layer:<2} {reason}")
|
||
abort(status)
|
||
@app.before_request
|
||
def csrf_gate():
|
||
if request.method in SAFE: # safe methods never change state (routing enforces it)
|
||
return
|
||
# layer 2a — browser-asserted provenance; page script cannot set this header
|
||
if request.headers.get("Sec-Fetch-Site") == "cross-site":
|
||
reject(403, "2a", "Sec-Fetch-Site: cross-site")
|
||
# layer 2b — exact-origin allow-list; catches sibling subdomains that are same-site
|
||
origin = request.headers.get("Origin")
|
||
if origin is None or origin not in ALLOWED_ORIGINS:
|
||
reject(403, "2b", f"Origin {origin} not in allow-list")
|
||
# layer 3 — session-bound synchronizer token, constant-time compare
|
||
expected = session.get("csrf", "")
|
||
supplied = request.form.get("csrf") or request.headers.get("X-CSRF-Token", "")
|
||
if not expected or not hmac.compare_digest(expected.encode(), supplied.encode()):
|
||
reject(403, "3", "token missing or mismatched")
|
||
def csrf_token(): # called by the template that renders the form
|
||
return session.setdefault("csrf", secrets.token_urlsafe(32))
|
||
@app.errorhandler(405)
|
||
def method_not_allowed(_e): # layer 0 — the route itself refuses GET
|
||
log.info(f"[GATE] REJECT 405 {request.method:<4} {request.full_path.rstrip('?'):<48} layer=0 method not allowed on state-changing route")
|
||
return "Method Not Allowed\n", 405
|
||
# ---- scaffolding for the experiment (login, account page, state) — not part of the gate ----
|
||
USERS = {"wiener": {"email": "[email protected]"}}
|
||
@app.get("/login")
|
||
def login():
|
||
session["user"] = "wiener"
|
||
csrf_token()
|
||
return "logged in as wiener\n"
|
||
@app.get("/my-account")
|
||
def my_account():
|
||
if "user" not in session:
|
||
abort(401)
|
||
return render_template_string(
|
||
'<form method="POST" action="/change-email">'
|
||
'<input type="hidden" name="csrf" value="{{ t }}">'
|
||
'<input name="email"><button>Update email</button></form>\n'
|
||
'current email: {{ e }}\n',
|
||
t=csrf_token(), e=USERS["wiener"]["email"])
|
||
@app.post("/change-email") # layer 0: POST only → GET receives 405 from Flask
|
||
def change_email():
|
||
if "user" not in session:
|
||
abort(401)
|
||
USERS[session["user"]]["email"] = request.form["email"]
|
||
log.info(f"[APP ] ACCEPT 200 POST /change-email{'':<36} layers 0,2a,2b,3 passed email={request.form['email']}")
|
||
return f"email changed to {request.form['email']}\n"
|
||
if __name__ == "__main__":
|
||
app.run(host="127.0.0.1", port=8443, ssl_context=("cert.pem", "key.pem")) |
+48
−0
Loading