XSRF & Cross-Origin Protection

import asyncio
import hashlib
import hmac
import json
import logging
import re
import secrets
from http.cookies import SimpleCookie
from urllib.parse import urlencode, urlsplit

from tornado import httpclient, httpserver, netutil, template, web
from tornado.escape import json_decode


class Capture(logging.Handler):
    """Collect the warnings Tornado logs for rejected requests."""

    def __init__(self):
        super().__init__()
        self.lines = []

    def emit(self, record):
        if record.name == "tornado.general":
            self.lines.append(record.getMessage().splitlines()[0])


captured = Capture()
tornado_log = logging.getLogger("tornado")
tornado_log.addHandler(captured)
tornado_log.propagate = False
tornado_log.setLevel(logging.INFO)


class Browser:
    """A server on a free loopback port plus a cookie jar, standing in for a browser."""

    def __init__(self, app):
        self.app = app
        self.cookies = {}

    async def __aenter__(self):
        self.server = httpserver.HTTPServer(self.app)
        sockets = netutil.bind_sockets(0, "127.0.0.1")
        self.server.add_sockets(sockets)
        self.base = f"http://127.0.0.1:{sockets[0].getsockname()[1]}"
        return self

    async def __aexit__(self, *exc_info):
        self.server.stop()
        await self.server.close_all_connections()

    async def request(self, method, path, headers=None, body=None, send_cookies=True):
        headers = dict(headers or {})
        if send_cookies and self.cookies:
            headers["Cookie"] = "; ".join(f"{k}={v}" for k, v in self.cookies.items())
        if body is None and method in ("POST", "PUT", "PATCH"):
            body = ""
        response = await httpclient.AsyncHTTPClient().fetch(
            self.base + path, method=method, headers=headers, body=body,
            raise_error=False, follow_redirects=False)
        for line in response.headers.get_list("Set-Cookie"):
            for name, morsel in SimpleCookie(line).items():
                self.cookies[name] = morsel.value
        return response


FORM = {"Content-Type": "application/x-www-form-urlencoded"}
SHELF_PAGE = """<h1>Bookshelf Live</h1>
<form action="/books" method="post">
  {% module xsrf_form_html() %}
  <input name="title"><button>Add</button>
</form>"""
BOOKS = []


class ShelfHandler(web.RequestHandler):
    def get(self):
        self.render("shelf.html")


class PlainHandler(web.RequestHandler):
    def get(self):
        self.write("no form here")


class BooksHandler(web.RequestHandler):
    def post(self):
        BOOKS.append(self.get_body_argument("title"))
        self.write({"books": BOOKS})


def make_app(**extra):
    settings = dict(
        template_loader=template.DictLoader({"shelf.html": SHELF_PAGE}),
        xsrf_cookies=True,
    )
    settings.update(extra)
    return web.Application([
        (r"/", ShelfHandler), (r"/plain", PlainHandler), (r"/books", BooksHandler),
    ], **settings)


async def main():
    async with Browser(make_app()) as browser:
        plain = await browser.request("GET", "/plain")
        print("GET /plain -> cookie set:", "Set-Cookie" in plain.headers)

        page = await browser.request("GET", "/")
        print("GET /      -> cookie set:", "_xsrf" in browser.cookies)
        field = re.search(r'name="_xsrf" value="([^"]+)"', page.body.decode()).group(1)

        forged = await browser.request("POST", "/books", FORM, urlencode({"title": "Forged"}))
        print("POST without token ->", forged.code)

        body = urlencode({"title": "Dune", "_xsrf": field})
        genuine = await browser.request("POST", "/books", FORM, body)
        print("POST with form token ->", genuine.code, json_decode(genuine.body))

        wrong = urlencode({"title": "Forged", "_xsrf": "2|00|00|1"})
        print("POST with a wrong token ->", (await browser.request("POST", "/books", FORM, wrong)).code)

        nocookie = await browser.request("POST", "/books", FORM, body, send_cookies=False)
        print("POST with token but no cookie ->", nocookie.code)
    for line in captured.lines:
        print("log:", line)


asyncio.run(main())

def unmask(value):
    version, mask, masked, timestamp = value.split("|")
    mask, masked = bytes.fromhex(mask), bytes.fromhex(masked)
    return bytes(b ^ mask[i % 4] for i, b in enumerate(masked)).hex(), version


async def main():
    async with Browser(make_app()) as browser:
        first = await browser.request("GET", "/")
        cookie = browser.cookies["_xsrf"]
        second = await browser.request("GET", "/")      # same cookie sent, new mask in the page
        fields = [re.search(r'value="([^"]+)"', r.body.decode()).group(1) for r in (first, second)]
        print("cookie equals first field:", cookie == fields[0])   # the cookie was set by that first response
        print("cookie differs from second field:", cookie != fields[1])
        print("fields differ from each other:", fields[0] != fields[1])
        print("same underlying token:", unmask(cookie)[0] == unmask(fields[0])[0] == unmask(fields[1])[0])
        print("version:", unmask(cookie)[1])


asyncio.run(main())

class ApiBooksHandler(web.RequestHandler):
    def get(self):
        self.write({"books": BOOKS, "xsrf": self.xsrf_token.decode()})   # reading the property sets the cookie

    def post(self):
        BOOKS.append(json_decode(self.request.body)["title"])
        self.set_status(201)
        self.write({"books": BOOKS})

    def delete(self):
        BOOKS.clear()
        self.set_status(204)


async def main():
    app = web.Application([(r"/api/books", ApiBooksHandler)], xsrf_cookies=True)
    async with Browser(app) as browser:
        token = json_decode((await browser.request("GET", "/api/books")).body)["xsrf"]
        payload = json.dumps({"title": "Hyperion"})
        JSON = {"Content-Type": "application/json"}

        r = await browser.request("POST", "/api/books", JSON, payload)
        print("JSON POST, no token:", r.code)
        r = await browser.request("POST", "/api/books", {**JSON, "X-XSRFToken": token}, payload)
        print("X-XSRFToken header:", r.code)
        r = await browser.request("POST", "/api/books", {**JSON, "X-CSRFToken": token}, payload)
        print("X-CSRFToken header:", r.code)
        r = await browser.request("POST", "/api/books?_xsrf=" + token, JSON, payload)
        print("_xsrf query argument:", r.code)
        r = await browser.request("DELETE", "/api/books")
        print("DELETE, no token:", r.code)
        r = await browser.request("DELETE", "/api/books", {"X-XSRFToken": token})
        print("DELETE with header:", r.code)


asyncio.run(main())

// token comes from <meta name="xsrf-token" content="..."> rendered by the server
const token = document.querySelector('meta[name="xsrf-token"]').content;

async function addBook(title) {
  const response = await fetch("/api/books", {
    method: "POST",
    credentials: "same-origin",
    headers: {"Content-Type": "application/json", "X-XSRFToken": token},
    body: JSON.stringify({title}),
  });
  if (response.status === 403) throw new Error("XSRF token rejected: reload the page");
  return response.json();
}

async def main():
    app = make_app(
        xsrf_cookie_name="__Host-xsrf",
        xsrf_cookie_kwargs={"secure": True, "httponly": True, "samesite": "Lax"},
    )
    async with Browser(app) as browser:
        page = await browser.request("GET", "/")
        print(page.headers["Set-Cookie"])
        field = re.search(r'value="([^"]+)"', page.body.decode()).group(1)
        body = urlencode({"title": "Foundation", "_xsrf": field})
        print("POST ->", (await browser.request("POST", "/books", FORM, body)).code)


asyncio.run(main())

WEBHOOK_KEY = secrets.token_bytes(32)      # shared with the sender; load it from your secret store


class WebhookHandler(web.RequestHandler):
    def check_xsrf_cookie(self):
        expected = hmac.new(WEBHOOK_KEY, self.request.body, hashlib.sha256).hexdigest()
        if not hmac.compare_digest(self.request.headers.get("X-Signature", ""), expected):
            raise web.HTTPError(403, "bad webhook signature")

    def post(self):
        self.write({"received": json_decode(self.request.body)})


async def main():
    app = web.Application([(r"/hooks/shelf", WebhookHandler)], xsrf_cookies=True)
    async with Browser(app) as browser:
        body = json.dumps({"event": "book.added"})
        signature = hmac.new(WEBHOOK_KEY, body.encode(), hashlib.sha256).hexdigest()
        ok = await browser.request("POST", "/hooks/shelf", {"X-Signature": signature}, body)
        bad = await browser.request("POST", "/hooks/shelf", {"X-Signature": "0" * 64}, body)
        print("signed:", ok.code, "forged signature:", bad.code)


asyncio.run(main())

API_TOKENS = {secrets.token_hex(16): "alice"}       # issued by your token endpoint in a real application


class ApiHandler(web.RequestHandler):
    def _bearer_user(self):
        scheme, _, token = self.request.headers.get("Authorization", "").partition(" ")
        return API_TOKENS.get(token) if scheme == "Bearer" else None

    def check_xsrf_cookie(self):
        if self._bearer_user() is None:          # no valid bearer token: full check, cookies in play
            super().check_xsrf_cookie()

    def get_current_user(self):
        return self._bearer_user()               # this handler never reads the session cookie

    def post(self):
        if not self.current_user:
            raise web.HTTPError(401)
        self.write({"added_by": self.current_user})


async def main():
    app = web.Application([(r"/api/shelf", ApiHandler)], xsrf_cookies=True)
    async with Browser(app) as browser:
        token = next(iter(API_TOKENS))
        r = await browser.request("POST", "/api/shelf", {"Authorization": f"Bearer {token}"})
        print("valid bearer token, no XSRF token:", r.code, json_decode(r.body))
        r = await browser.request("POST", "/api/shelf", {"Authorization": "Bearer not-a-token"})
        print("invalid bearer token:", r.code)
        r = await browser.request("POST", "/api/shelf")
        print("no credentials:", r.code)


asyncio.run(main())

TRUSTED_ORIGINS = {"https://app.example.com"}


class FetchMetadataMixin:
    """Reject unsafe requests that browsers mark as cross-site. Put it first in the base classes."""

    def prepare(self):
        if self.request.method in ("GET", "HEAD", "OPTIONS"):
            return
        site = self.request.headers.get("Sec-Fetch-Site")
        origin = self.request.headers.get("Origin")
        if site in ("same-origin", "none"):
            return
        if site is None and origin is None:
            return                                  # old browser or non-browser client
        if site is None and urlsplit(origin).netloc.lower() == self.request.host.lower():
            return                                  # no Fetch Metadata: compare Origin and Host
        if origin in TRUSTED_ORIGINS:
            return
        raise web.HTTPError(403, "cross-origin request %s from %s", site, origin)


class GuardedBooksHandler(FetchMetadataMixin, web.RequestHandler):
    def post(self):
        self.write({"added": self.get_body_argument("title")})


async def main():
    app = web.Application([(r"/books", GuardedBooksHandler)])      # no xsrf_cookies at all
    async with Browser(app) as browser:
        host = browser.base.removeprefix("http://")
        cases = {
            "non-browser client": {},
            "same-origin": {"Sec-Fetch-Site": "same-origin"},
            "cross-site": {"Sec-Fetch-Site": "cross-site", "Origin": "https://evil.example"},
            "cross-site, trusted origin": {"Sec-Fetch-Site": "cross-site", "Origin": "https://app.example.com"},
            "no Sec-Fetch-Site, Origin equals Host": {"Origin": f"http://{host}"},
            "no Sec-Fetch-Site, foreign Origin": {"Origin": "https://evil.example"},
        }
        for name, headers in cases.items():
            r = await browser.request("POST", "/books", {**FORM, **headers}, "title=Emma")
            print(f"{name}: {r.code}")


asyncio.run(main())

# norun - needs Tornado 6.6, not released at the time of writing
import urllib.parse

from tornado import web


class BookshelfHandler(web.RequestHandler):
    def check_origin(self, origin):
        # allow any sub-domain of our own site, in addition to the trusted_origins list
        netloc = urllib.parse.urlsplit(origin).netloc
        return super().check_origin(origin) or netloc.endswith(".bookshelf.example")


application = web.Application(
    [(r"/books", BookshelfHandler)],
    cross_origin_protection=True,
    trusted_origins=["https://app.bookshelf.example"],
)

from tornado import websocket

ALLOWED_WS_ORIGINS = {"https://app.example.com"}


class LiveShelfSocket(websocket.WebSocketHandler):
    def check_origin(self, origin):
        return origin in ALLOWED_WS_ORIGINS or super().check_origin(origin)

    def open(self):
        self.write_message("connected")


async def main():
    app = web.Application([(r"/live", LiveShelfSocket)])
    async with Browser(app) as browser:
        url = browser.base.replace("http", "ws") + "/live"
        host_origin = browser.base
        for origin in (None, host_origin, "https://app.example.com", "https://evil.example"):
            headers = {"Origin": origin} if origin else {}
            try:
                connection = await websocket.websocket_connect(httpclient.HTTPRequest(url, headers=headers))
                print(f"Origin {origin}: accepted, first message {await connection.read_message()!r}")
                connection.close()
            except httpclient.HTTPClientError as error:
                print(f"Origin {origin}: refused with {error.code}")


asyncio.run(main())

CORS_ORIGINS = {"https://app.example.com"}


class CorsMixin:
    def set_default_headers(self):
        self.set_header("Vary", "Origin")
        origin = self.request.headers.get("Origin")
        if origin in CORS_ORIGINS:
            self.set_header("Access-Control-Allow-Origin", origin)
            self.set_header("Access-Control-Allow-Credentials", "true")

    def options(self, *args, **kwargs):               # the preflight
        self.set_header("Access-Control-Allow-Methods", "GET, POST, DELETE")
        self.set_header("Access-Control-Allow-Headers", "Content-Type, X-XSRFToken")
        self.set_header("Access-Control-Max-Age", "600")
        self.set_status(204)


class CorsBooksHandler(CorsMixin, web.RequestHandler):
    def get(self):
        self.write({"books": ["Dune"]})


async def main():
    app = web.Application([(r"/api/books", CorsBooksHandler)], xsrf_cookies=True)
    async with Browser(app) as browser:
        for origin in ("https://app.example.com", "https://evil.example"):
            pre = await browser.request("OPTIONS", "/api/books", {"Origin": origin})
            get = await browser.request("GET", "/api/books", {"Origin": origin})
            print(origin, "->", pre.code, get.headers.get("Access-Control-Allow-Origin"), get.headers.get("Vary"))


asyncio.run(main())

class SecureBaseHandler(web.RequestHandler):
    def set_default_headers(self):
        self.set_header("Content-Security-Policy", "default-src 'self'; frame-ancestors 'none'")
        self.set_header("X-Content-Type-Options", "nosniff")
        self.set_header("Referrer-Policy", "same-origin")
        self.set_header("Cache-Control", "no-store")


class SessionHandler(SecureBaseHandler):
    def get(self):
        self.set_signed_cookie("session", "alice", httponly=True, secure=True, samesite="Lax")
        self.write("signed in")


async def main():
    app = web.Application([(r"/session", SessionHandler)], cookie_secret=secrets.token_hex(32))
    async with Browser(app) as browser:
        r = await browser.request("GET", "/session")
        for name in ("Content-Security-Policy", "X-Content-Type-Options", "Referrer-Policy", "Cache-Control"):
            print(f"{name}: {r.headers[name]}")
        attributes = r.headers["Set-Cookie"].split("; ")[1:]
        print("cookie attributes:", [a for a in attributes if a.lower() in ("secure", "httponly", "samesite=lax")])


asyncio.run(main())

import unittest

from tornado.testing import AsyncHTTPTestCase, ExpectLog


class XsrfProtectionTest(AsyncHTTPTestCase):
    def get_app(self):
        return make_app()

    def token_and_cookie(self):
        response = self.fetch("/")
        field = re.search(r'name="_xsrf" value="([^"]+)"', response.body.decode()).group(1)
        return field, response.headers["Set-Cookie"].split(";")[0]

    def test_post_without_token_is_forbidden(self):
        with ExpectLog("tornado.general", ".*'_xsrf' argument missing from POST"):
            response = self.fetch("/books", method="POST", body="title=Dune")
        self.assertEqual(response.code, 403)

    def test_post_with_cookie_and_token_succeeds(self):
        field, cookie = self.token_and_cookie()
        response = self.fetch("/books", method="POST", body=urlencode({"title": "Dune", "_xsrf": field}),
                              headers={"Cookie": cookie})
        self.assertEqual(response.code, 200)

    def test_header_token_is_accepted(self):
        field, cookie = self.token_and_cookie()
        response = self.fetch("/books", method="POST", body="title=Dune",
                              headers={"Cookie": cookie, "X-XSRFToken": field})
        self.assertEqual(response.code, 200)

    def test_token_without_cookie_is_forbidden(self):
        field, _ = self.token_and_cookie()
        with ExpectLog("tornado.general", ".*XSRF cookie does not match POST argument"):
            response = self.fetch("/books", method="POST", body=urlencode({"title": "Dune", "_xsrf": field}))
        self.assertEqual(response.code, 403)


outcome = unittest.TextTestRunner(verbosity=0).run(unittest.defaultTestLoader.loadTestsFromTestCase(XsrfProtectionTest))
print("tests run:", outcome.testsRun, "successful:", outcome.wasSuccessful())