#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Ginkgo (DUFE) sign-in code query service. Endpoint: GET|POST /dufe/ginkgostu_sign_in params: username, password(base64), url(base64) [commit=true] optional, default true: also POST the code back Listen: 127.0.0.1:9005 (WSGI: waitress) """ import base64 import binascii import json import logging import re import sys import urllib.parse import uuid try: import requests except ImportError: sys.exit("Missing dependency, run: pip install requests") try: from flask import Flask, request, jsonify except ImportError: sys.exit("Missing dependency, run: pip install flask") try: from waitress import serve except ImportError: sys.exit("Missing dependency, run: pip install waitress") requests.packages.urllib3.disable_warnings() # ------------------------------------------------------------------------------ # Site constants # ------------------------------------------------------------------------------ GINKGO = "https://ginkgostu.dufe.edu.cn" SSO = "https://sso.dufe.edu.cn" CUSTOM_CODE = "kb_dconline" UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36") logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", ) log = logging.getLogger("ginkgo") # ------------------------------------------------------------------------------ # Error taxonomy # AuthError -> 401 caller-side credential / captcha problem # NoSignInCode -> 404 activity exists but has no code configured # UpstreamError -> 502 remote service misbehaved / changed shape # anything else -> 500 internal, generic message to the client # ------------------------------------------------------------------------------ class AuthError(Exception): """Authentication failed (bad credentials, captcha rejected, locked account).""" class NoSignInCodeError(Exception): """The sign-in activity exists but has no code configured.""" class UpstreamError(Exception): """Remote service returned an unexpected or erroneous response.""" # ============================================================================== # Part 1: Pure-python AES-128 # ============================================================================== _SBOX = [ 0x63,0x7c,0x77,0x7b,0xf2,0x6b,0x6f,0xc5,0x30,0x01,0x67,0x2b,0xfe,0xd7,0xab,0x76, 0xca,0x82,0xc9,0x7d,0xfa,0x59,0x47,0xf0,0xad,0xd4,0xa2,0xaf,0x9c,0xa4,0x72,0xc0, 0xb7,0xfd,0x93,0x26,0x36,0x3f,0xf7,0xcc,0x34,0xa5,0xe5,0xf1,0x71,0xd8,0x31,0x15, 0x04,0xc7,0x23,0xc3,0x18,0x96,0x05,0x9a,0x07,0x12,0x80,0xe2,0xeb,0x27,0xb2,0x75, 0x09,0x83,0x2c,0x1a,0x1b,0x6e,0x5a,0xa0,0x52,0x3b,0xd6,0xb3,0x29,0xe3,0x2f,0x84, 0x53,0xd1,0x00,0xed,0x20,0xfc,0xb1,0x5b,0x6a,0xcb,0xbe,0x39,0x4a,0x4c,0x58,0xcf, 0xd0,0xef,0xaa,0xfb,0x43,0x4d,0x33,0x85,0x45,0xf9,0x02,0x7f,0x50,0x3c,0x9f,0xa8, 0x51,0xa3,0x40,0x8f,0x92,0x9d,0x38,0xf5,0xbc,0xb6,0xda,0x21,0x10,0xff,0xf3,0xd2, 0xcd,0x0c,0x13,0xec,0x5f,0x97,0x44,0x17,0xc4,0xa7,0x7e,0x3d,0x64,0x5d,0x19,0x73, 0x60,0x81,0x4f,0xdc,0x22,0x2a,0x90,0x88,0x46,0xee,0xb8,0x14,0xde,0x5e,0x0b,0xdb, 0xe0,0x32,0x3a,0x0a,0x49,0x06,0x24,0x5c,0xc2,0xd3,0xac,0x62,0x91,0x95,0xe4,0x79, 0xe7,0xc8,0x37,0x6d,0x8d,0xd5,0x4e,0xa9,0x6c,0x56,0xf4,0xea,0x65,0x7a,0xae,0x08, 0xba,0x78,0x25,0x2e,0x1c,0xa6,0xb4,0xc6,0xe8,0xdd,0x74,0x1f,0x4b,0xbd,0x8b,0x8a, 0x70,0x3e,0xb5,0x66,0x48,0x03,0xf6,0x0e,0x61,0x35,0x57,0xb9,0x86,0xc1,0x1d,0x9e, 0xe1,0xf8,0x98,0x11,0x69,0xd9,0x8e,0x94,0x9b,0x1e,0x87,0xe9,0xce,0x55,0x28,0xdf, 0x8c,0xa1,0x89,0x0d,0xbf,0xe6,0x42,0x68,0x41,0x99,0x2d,0x0f,0xb0,0x54,0xbb,0x16, ] _INV_SBOX = [0] * 256 for _i, _v in enumerate(_SBOX): _INV_SBOX[_v] = _i _RCON = [0x01, 0x02, 0x04, 0x08, 0x10, 0x20, 0x40, 0x80, 0x1B, 0x36] def _gmul(a, b): p = 0 for _ in range(8): if b & 1: p ^= a hi = a & 0x80 a = (a << 1) & 0xFF if hi: a ^= 0x1B b >>= 1 return p def _expand_key(key): w = [list(key[4 * i:4 * i + 4]) for i in range(4)] for i in range(4, 44): t = list(w[i - 1]) if i % 4 == 0: t = t[1:] + t[:1] t = [_SBOX[b] for b in t] t[0] ^= _RCON[i // 4 - 1] w.append([w[i - 4][j] ^ t[j] for j in range(4)]) return w def _add_round_key(s, w, rnd): for c in range(4): for r in range(4): s[r][c] ^= w[rnd * 4 + c][r] def _sub_bytes(s, box): for r in range(4): for c in range(4): s[r][c] = box[s[r][c]] def _shift_rows(s): for r in range(1, 4): s[r] = s[r][r:] + s[r][:r] def _inv_shift_rows(s): for r in range(1, 4): s[r] = s[r][-r:] + s[r][:-r] def _mix_columns(s): for c in range(4): a = [s[r][c] for r in range(4)] s[0][c] = _gmul(a[0], 2) ^ _gmul(a[1], 3) ^ a[2] ^ a[3] s[1][c] = a[0] ^ _gmul(a[1], 2) ^ _gmul(a[2], 3) ^ a[3] s[2][c] = a[0] ^ a[1] ^ _gmul(a[2], 2) ^ _gmul(a[3], 3) s[3][c] = _gmul(a[0], 3) ^ a[1] ^ a[2] ^ _gmul(a[3], 2) def _inv_mix_columns(s): for c in range(4): a = [s[r][c] for r in range(4)] s[0][c] = _gmul(a[0],14) ^ _gmul(a[1],11) ^ _gmul(a[2],13) ^ _gmul(a[3],9) s[1][c] = _gmul(a[0],9) ^ _gmul(a[1],14) ^ _gmul(a[2],11) ^ _gmul(a[3],13) s[2][c] = _gmul(a[0],13) ^ _gmul(a[1],9) ^ _gmul(a[2],14) ^ _gmul(a[3],11) s[3][c] = _gmul(a[0],11) ^ _gmul(a[1],13) ^ _gmul(a[2],9) ^ _gmul(a[3],14) class AES128: def __init__(self, key): if len(key) != 16: raise ValueError("AES-128 requires a 16-byte key, got %d" % len(key)) self.w = _expand_key(key) def encrypt_block(self, block): s = [[block[r + 4 * c] for c in range(4)] for r in range(4)] _add_round_key(s, self.w, 0) for rnd in range(1, 10): _sub_bytes(s, _SBOX) _shift_rows(s) _mix_columns(s) _add_round_key(s, self.w, rnd) _sub_bytes(s, _SBOX) _shift_rows(s) _add_round_key(s, self.w, 10) return bytes(s[r][c] for c in range(4) for r in range(4)) def decrypt_block(self, block): s = [[block[r + 4 * c] for c in range(4)] for r in range(4)] _add_round_key(s, self.w, 10) for rnd in range(9, 0, -1): _inv_shift_rows(s) _sub_bytes(s, _INV_SBOX) _add_round_key(s, self.w, rnd) _inv_mix_columns(s) _inv_shift_rows(s) _sub_bytes(s, _INV_SBOX) _add_round_key(s, self.w, 0) return bytes(s[r][c] for c in range(4) for r in range(4)) def _pkcs7_unpad(data): if not data: return data n = data[-1] if 1 <= n <= 16 and data[-n:] == bytes([n]) * n: return data[:-n] return data def _zero_pad(data): if len(data) % 16: data += b"\x00" * (16 - len(data) % 16) return data def aes_ecb_decrypt_b64(cipher_b64, key_text): aes = AES128(key_text[:16].encode("utf-8")) raw = base64.b64decode(cipher_b64) out = b"".join(aes.decrypt_block(raw[i:i + 16]) for i in range(0, len(raw), 16)) return _pkcs7_unpad(out) def aes_cbc_encrypt_b64_zero(plain, key_text): key = key_text[:16].encode("utf-8") aes = AES128(key) iv = key data = _zero_pad(plain.encode("utf-8")) out, prev = b"", iv for i in range(0, len(data), 16): blk = bytes(x ^ y for x, y in zip(data[i:i + 16], prev)) prev = aes.encrypt_block(blk) out += prev return base64.b64encode(out).decode("ascii") # ============================================================================== # Part 2: Session & URL parsing # ============================================================================== def new_session(): s = requests.Session() s.trust_env = False s.verify = False s.headers.update({"User-Agent": UA, "Accept-Language": "zh-CN,zh;q=0.9"}) return s def parse_course_url(url): m = re.search( r"/course/detail/(?P\d+)/(?P[^/]+)/" r"(?P\d+)/signin/(?P\d+)/(?P\d+)", url) if not m: raise ValueError( "URL is not a sign-in page (missing /signin//): %s" % url) return m.groupdict() # ============================================================================== # Part 3: Login flow # ============================================================================== def get_unify_config(sess): r = sess.get(GINKGO + "/student/getLoginUnifySet", timeout=20) r.raise_for_status() try: return r.json()["data"] except (ValueError, KeyError) as e: raise UpstreamError("Malformed getLoginUnifySet response: %s" % e) def open_authorize(sess, cfg): params = { "response_type": cfg["responseType"], "client_id": cfg["clientId"], "redirect_uri": cfg["rollBackUrl"], "scope": cfg.get("scope", "") or "", "state": cfg.get("state", "") or "", } r = sess.get(cfg["loginUrl"], params=params, timeout=20, allow_redirects=True) r.raise_for_status() m = re.search(r'name="__token__"\s+value="([^"]+)"', r.text) if not m: raise UpstreamError("CSRF token __token__ not found on login page") return m.group(1), r.url def exploit_slider(sess, referer): url = (SSO + "/auth/widget?layer=image_verify&widget=Slider" "&action=get_verify&type=login_image_verify") r = sess.post(url, data={"width": 280, "height": 155, "block_size": 80}, headers={"X-Requested-With": "XMLHttpRequest", "Referer": referer}, timeout=20) r.raise_for_status() try: d = r.json()["data"] token, cipher = d["token"], d["d"] except (ValueError, KeyError, TypeError) as e: raise UpstreamError("Malformed slider challenge response: %s" % e) x = aes_ecb_decrypt_b64(cipher, token).decode("utf-8", "replace") return {"token": token, "x": x, "cipher": cipher} def submit_login(sess, post_url, csrf, username, password, slider, referer): form = { "fingerprint": uuid.uuid4().hex, "username": username, "password": aes_cbc_encrypt_b64_zero(password, csrf), "verify_token": slider["token"], "verify_code": slider["x"], "__token__": csrf, } return sess.post(post_url, data=form, timeout=25, allow_redirects=False, headers={"Referer": referer, "Origin": SSO, "Content-Type": "application/x-www-form-urlencoded"}) def exchange_token(sess, cfg, code): r = sess.post(GINKGO + "/student/authLogin", data={"code": code}, timeout=20, headers={"Content-Type": "application/x-www-form-urlencoded"}) try: j = r.json() except ValueError as e: raise UpstreamError("authLogin returned non-JSON: %s" % e) if j.get("code") != "2000": raise AuthError("OAuth code exchange refused: %s" % json.dumps(j, ensure_ascii=False)) return j["data"]["token"] # ============================================================================== # Part 4: Sign-in API # ============================================================================== def api_headers(token, fingerprint=None): return { "User-Agent": UA, "Origin": GINKGO, "Referer": GINKGO + "/", "Content-Type": "application/x-www-form-urlencoded", "token": token, "fingerprint": fingerprint or uuid.uuid4().hex, } def call_api(sess, path, data, token): return sess.post(GINKGO + path, data=data, timeout=20, headers=api_headers(token)) def commit_sign_in(sess, token, sign_in_id, batch_no, referer, code): """ POST /student/signIn/InTheSign to actually check in with the given code. Mirrors the browser request: headers: batchno, fingerprint, token, Referer (course page) body: signInId=&code= Returns (ok: bool, message: str, raw: dict|None). """ headers = api_headers(token) if batch_no: # NOTE: the browser sends a "batchno" header that does not always equal # the batchNo segment of the course URL. We pass the URL segment here; # the server appears not to enforce it. headers["batchno"] = str(batch_no) headers["Referer"] = referer try: r = sess.post( GINKGO + "/student/signIn/InTheSign", data={"signInId": str(sign_in_id), "code": str(code)}, timeout=20, headers=headers, ) except requests.RequestException as e: return False, "request to InTheSign failed: %s" % e, None if r.status_code != 200: return False, "InTheSign returned HTTP %d" % r.status_code, None try: j = r.json() except ValueError: return False, "InTheSign returned non-JSON: %s" % r.text[:200], None api_code = str(j.get("code", "")) server_msg = (j.get("message") or j.get("msg") or "").strip() if api_code == "2000": return True, server_msg or "accepted", j if server_msg: reason = server_msg elif api_code: reason = "upstream returned code %s" % api_code else: reason = ("unrecognized upstream response: %s" % json.dumps(j, ensure_ascii=False)[:200]) return False, reason, j def fetch_sign_in_info(username, password, course_url, do_commit=True): route = parse_course_url(course_url) sign_in_id = route["signInId"] batch_no = route.get("batchNo") sess = new_session() cfg = get_unify_config(sess) csrf, page_url = open_authorize(sess, cfg) slider = exploit_slider(sess, page_url) resp = submit_login(sess, page_url, csrf, username, password, slider, page_url) loc = resp.headers.get("Location", "") m = re.search(r"[?&]code=([A-Za-z0-9_\-]+)", loc) if not m: raise AuthError( "Login failed: no OAuth code in redirect (HTTP %s)" % resp.status_code) auth_code = m.group(1) token = exchange_token(sess, cfg, auth_code) r = call_api(sess, "/student/signIn/getSignInInfo", {"signInId": sign_in_id}, token) try: j = r.json() except ValueError as e: raise UpstreamError("getSignInInfo returned non-JSON: %s" % e) data = j.get("data") or {} model = data.get("signInModel") or {} code = model.get("code") if code is None or code == "": raise NoSignInCodeError( "Sign-in activity %s has no code configured" % sign_in_id) result = { "signInId": sign_in_id, "signInCode": code, "type": model.get("type"), "title": model.get("title"), "status": data.get("signStatus"), } if do_commit: ok, reason, _raw = commit_sign_in( sess, token, sign_in_id, batch_no, course_url, code ) result["commit"] = {"success": ok, "message": reason} return result # ============================================================================== # Part 5: HTTP service (WSGI app) # ============================================================================== app = Flask(__name__) def _b64_decode(value, field): """ Decode a base64 query/form parameter tolerantly. Handles, in order: * '%3D' / '%2B' / '%20' -- iOS Shortcuts and some HTTP clients URL-encode the base64 payload (notably the trailing '='), so we percent-decode first. * '+' turned into ' ' by application/x-www-form-urlencoded. * trailing newline from `base64` CLI without -w0. * both standard (+/) and urlsafe (-_) alphabets. """ v = urllib.parse.unquote(value or "") v = v.strip() v = v.replace(" ", "+") v = v.replace("\r", "").replace("\n", "") if not v: raise ValueError("Empty base64 for '%s'" % field) try: return base64.b64decode(v, altchars=b"-_", validate=True).decode("utf-8") except (binascii.Error, UnicodeDecodeError, ValueError) as e: raise ValueError("Invalid base64 for '%s': %s" % (field, e)) def _error(http_status, code, msg): return jsonify({"code": code, "msg": msg, "data": None}), http_status @app.route("/dufe/ginkgostu_sign_in", methods=["GET", "POST"]) def ginkgostu_sign_in(): args = request.values username = (args.get("username") or "").strip() password_b64 = args.get("password") or "" url_b64 = args.get("url") or "" commit_arg = (args.get("commit") or "true").strip().lower() do_commit = commit_arg not in ("0", "false", "no") if not (username and password_b64 and url_b64): return _error(400, 400, "Missing parameters: username, password(base64), url(base64)") try: password = _b64_decode(password_b64, "password") course_url = _b64_decode(url_b64, "url") except ValueError as e: return _error(400, 400, str(e)) log.info("query start user=%s commit=%s", username[:2] + "***", do_commit) try: data = fetch_sign_in_info(username, password, course_url, do_commit=do_commit) if do_commit: # Fail-closed: only an explicit success=true counts as signed in. commit_info = data.get("commit") if not isinstance(commit_info, dict) or commit_info.get("success") is not True: reason = (commit_info.get("message") if isinstance(commit_info, dict) else None) or "no commit result returned" log.warning("sign-in commit failed user=%s reason=%s", username[:2] + "***", reason) return jsonify({ "code": 200, "msg": "sign-in failed: %s" % reason, "data": data, }) log.info("query ok user=%s code=%s", username[:2] + "***", data.get("signInCode")) return jsonify({"code": 200, "msg": "ok", "data": data}) except AuthError as e: log.warning("auth failed: %s", e) return _error(401, 401, str(e)) except NoSignInCodeError as e: log.info("no code: %s", e) return _error(404, 404, str(e)) except UpstreamError as e: log.error("upstream error: %s", e) return _error(502, 502, str(e)) except ValueError as e: log.warning("bad request: %s", e) return _error(400, 400, str(e)) except requests.RequestException as e: log.error("upstream request failed: %s", e) return _error(502, 502, "Upstream request failed") except Exception: log.exception("internal error") return _error(500, 500, "Internal error") @app.route("/healthz", methods=["GET"]) def healthz(): return jsonify({"code": 200, "msg": "ok", "data": None}) # ============================================================================== # Part 6: WSGI entrypoint # ============================================================================== HOST = "127.0.0.1" PORT = 9005 THREADS = 16 BACKLOG = 128 CONNECTION_LIMIT = 256 CHANNEL_TIMEOUT = 120 def main(): log.info("serving on http://%s:%d (threads=%d)", HOST, PORT, THREADS) serve( app, host=HOST, port=PORT, threads=THREADS, backlog=BACKLOG, connection_limit=CONNECTION_LIMIT, channel_timeout=CHANNEL_TIMEOUT, ident="ginkgo-signin", ) return 0 if __name__ == "__main__": sys.exit(main())