130 lines
3.9 KiB
Python
130 lines
3.9 KiB
Python
import os, django
|
|
|
|
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")
|
|
django.setup()
|
|
|
|
import urllib.request, urllib.parse, urllib.error, json, base64, time
|
|
from apps.identity.models import User
|
|
from apps.application.models import Application, ApplicationStatus
|
|
from apps.oauth.models import OAuthScope
|
|
from apps.common.utils import generate_client_secret, hash_token
|
|
|
|
user, _ = User.objects.get_or_create(
|
|
email="e2e%d@example.com" % int(time.time()), defaults={"status": "active"}
|
|
)
|
|
user.set_password("S3cure-Pass-123")
|
|
user.status = "active"
|
|
user.email_verified = True
|
|
user.save()
|
|
|
|
sc, _ = OAuthScope.objects.get_or_create(code="openid", defaults={"is_system": True})
|
|
sp, _ = OAuthScope.objects.get_or_create(code="profile")
|
|
se, _ = OAuthScope.objects.get_or_create(code="email")
|
|
sph, _ = OAuthScope.objects.get_or_create(code="phone")
|
|
|
|
secret = generate_client_secret()
|
|
app, _ = Application.objects.get_or_create(
|
|
product_key="e2e",
|
|
defaults={
|
|
"name": "E2E",
|
|
"client_secret_hash": hash_token(secret),
|
|
"redirect_uris": ["https://e2e.com/cb"],
|
|
"grant_types": ["authorization_code", "refresh_token"],
|
|
"response_types": ["code"],
|
|
"status": ApplicationStatus.ACTIVE,
|
|
},
|
|
)
|
|
app.client_secret_hash = hash_token(secret)
|
|
app.scopes.set([sc, sp, se, sph])
|
|
app.save()
|
|
|
|
|
|
def post(path, data, headers=None):
|
|
req = urllib.request.Request(
|
|
"http://localhost:8000" + path,
|
|
data=urllib.parse.urlencode(data).encode(),
|
|
headers=headers or {},
|
|
)
|
|
try:
|
|
r = urllib.request.urlopen(req, timeout=10)
|
|
return r.status, r.read().decode()
|
|
except urllib.error.HTTPError as e:
|
|
return e.code, e.read().decode()
|
|
|
|
|
|
st, body = post(
|
|
"/api/v1/auth/login/", {"email": user.email, "password": "S3cure-Pass-123"}
|
|
)
|
|
print("login ->", st, body[:300])
|
|
tok = json.loads(body).get("access")
|
|
if not tok:
|
|
print("LOGIN BODY NO ACCESS:", body)
|
|
raise SystemExit(1)
|
|
auth_hdr = {"Authorization": f"Bearer {tok}"}
|
|
|
|
q = urllib.parse.urlencode(
|
|
{
|
|
"client_id": app.client_id,
|
|
"response_type": "code",
|
|
"redirect_uri": "https://e2e.com/cb",
|
|
"scope": "openid profile email phone",
|
|
"state": "x",
|
|
}
|
|
)
|
|
|
|
|
|
class NoRedirect(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
|
return None
|
|
|
|
|
|
opener = urllib.request.build_opener(NoRedirect)
|
|
req = urllib.request.Request(
|
|
"http://localhost:8000/oauth/authorize?" + q, headers=auth_hdr
|
|
)
|
|
try:
|
|
r = opener.open(req, timeout=10)
|
|
loc = r.headers.get("Location")
|
|
except urllib.error.HTTPError as e:
|
|
print("authorize HTTPError", e.code, e.read().decode()[:300])
|
|
loc = e.headers.get("Location")
|
|
print("authorize ->", loc)
|
|
code = urllib.parse.parse_qs(urllib.parse.urlparse(loc).query).get("code")
|
|
if not code:
|
|
print("NO CODE IN LOCATION; body of error if any above")
|
|
raise SystemExit(1)
|
|
code = code[0]
|
|
|
|
st2, body2 = post(
|
|
"/oauth/token",
|
|
{
|
|
"grant_type": "authorization_code",
|
|
"code": code,
|
|
"redirect_uri": "https://e2e.com/cb",
|
|
"client_id": app.client_id,
|
|
"client_secret": secret,
|
|
},
|
|
)
|
|
print("token ->", st2, body2[:300])
|
|
data = json.loads(body2) if body2 else {}
|
|
print("has id_token:", "id_token" in data)
|
|
_, p, _ = data["id_token"].split(".")
|
|
p += "=" * (-len(p) % 4)
|
|
claims = json.loads(base64.urlsafe_b64decode(p))
|
|
print(
|
|
"id_token claims: sub=%s email=%s iss=%s aud=%s"
|
|
% (claims.get("sub"), claims.get("email"), claims.get("iss"), claims.get("aud"))
|
|
)
|
|
|
|
# userinfo with the OAuth access token (carries openid profile email phone)
|
|
oauth_hdr = {"Authorization": f"Bearer {data['access_token']}"}
|
|
req2 = urllib.request.Request("http://localhost:8000/oauth/userinfo", headers=oauth_hdr)
|
|
r2 = urllib.request.urlopen(req2, timeout=10)
|
|
claims2 = json.loads(r2.read())
|
|
print(
|
|
"userinfo ->",
|
|
r2.status,
|
|
"email=%s" % claims2.get("email"),
|
|
"name=%s" % claims2.get("name"),
|
|
)
|