From ec439202078b999e4ebf37008a6545eceffdaae5 Mon Sep 17 00:00:00 2001 From: RRRosmontis Date: Thu, 10 Sep 2026 01:53:49 +0800 Subject: [PATCH] tmp: default protocol ss --- tmp-rrpp-deploy-20260910/app.py | 356 ++++++++++++++++++++++++++++++++ 1 file changed, 356 insertions(+) create mode 100644 tmp-rrpp-deploy-20260910/app.py diff --git a/tmp-rrpp-deploy-20260910/app.py b/tmp-rrpp-deploy-20260910/app.py new file mode 100644 index 0000000..d57c470 --- /dev/null +++ b/tmp-rrpp-deploy-20260910/app.py @@ -0,0 +1,356 @@ +# -*- coding: utf-8 -*- +""" +RRPP 订阅后端（api.rrrosmontis.icu）— ROSMANKA ID SSO 国家分层版 + +URL 形式： + /v1/rrpp/subscription?method=&user=&password=[&prefer=] + /v1/rrpp/nodes_update?user=...&password=... （旧端点，等价于 method=rrpp_client） + +数据结构（nodes.json，顶层为国家列表）： + [ + { + "country": "JP", + "available_level": [1, 2, 3], + "provider": "Azure", + "traffic": "15G", + "domain": "iam.aeolianchi.me", + "methods": { + "http": { "port": ..., "username": ..., "password": ... }, + "vless": { "port": ..., "uuid": ..., "flow": ..., "sni": ..., "public_key": ..., "short_id": ... }, + "ss": { "port": ..., "method": ..., "password": ... } + } + }, + ... + ] + +协议选择（clash / v2ray）： + prefer 参数指定优先协议（vless / ss / http），缺省默认 vless（最好用）。 + 若该国家无 prefer 协议，按 vless → ss → http 回退。 + +method 行为： + rrpp_client → 仅用 methods.http，返回旧格式节点 JSON（domain/port/country/username/password/remark）， + remark = " "（IPv6 加 "(IPv6)"），保证自建客户端兼容性。 + clash → 按协议生成 Clash YAML，备注 "R.R.P.P. - <国家代码>"（IPv6 加 "(IPv6)"） + v2ray → 按协议生成 base64 节点串（vless:// / ss:// / http://），同备注 + +tier 过滤：国家对象 available_level 包含用户 tier 才可见。 +tier=0 → 403；未知 method → 404；nodes.json 缺失/损坏 → 503。 + +运行：python app.py（监听 127.0.0.1:5010，nginx 反向代理） +服务：由 nssm 注册为 Windows 服务 RRPPSubBackend，开机自启 +""" +import base64 +import json +import urllib.error +import urllib.parse +import urllib.request + +from flask import Flask, request, Response + +app = Flask(__name__) + +SSO_API = 'http://127.0.0.1:4550/v1/sso' +SSO_SECRETS_PATH = 'C:/BtSoft/nginx/others/backEnd/sso-rrrosmontis-icu/secrets.json' +NODES_JSON_PATH = 'C:/BtSoft/nginx/others/htmlResources/private/RRPP/nodes.json' + +# 协议优先级（缺省时按此顺序回退） +PROTOCOL_PRIORITY = ['ss', 'http', 'vless'] + + +def _service_token(): + """读取本机 SSO 的服务共享密钥（单一密钥源，不落代码）。""" + try: + with open(SSO_SECRETS_PATH, 'r', encoding='utf-8') as f: + return json.load(f).get('service_tokens', {}).get('rrpp', '') + except Exception: + return '' + + +def b64url_decode(s): + """URL-safe base64 解码（容忍无 padding 与标准 base64）。""" + if not s: + return None + pad = '=' * (-len(s) % 4) + try: + return base64.urlsafe_b64decode((s + pad).encode('ascii')).decode('utf-8') + except Exception: + return None + + +def fetch_user_tier(email, password, client_ip=''): + """SSO 校验邮箱+密码并取 rrpp tier。返回 (ok, err, tier)。 + client_ip 透传给 SSO（X-Real-IP），使 SSO 限流按真实客户端 IP 计。""" + try: + req = urllib.request.Request( + SSO_API + '/service/rrpp/nodes', + data=json.dumps({'email': email, 'password': password}).encode('utf-8'), + headers={'Content-Type': 'application/json', + 'X-SSO-Service-Token': _service_token(), + 'X-Real-IP': client_ip}) + with urllib.request.urlopen(req, timeout=10) as resp: + d = json.loads(resp.read()) + return True, '', d.get('tier', 0) + except urllib.error.HTTPError as e: + try: + msg = json.loads(e.read()).get('error', '认证失败') + except Exception: + msg = '认证失败' + return False, msg, 0 + except Exception: + return False, 'SSO 服务不可达', 0 + + +def load_countries(): + """读 nodes.json（国家分层）。返回国家列表，失败抛异常。""" + with open(NODES_JSON_PATH, 'r', encoding='utf-8') as f: + return json.load(f) + + +def filter_countries_by_tier(countries, tier): + """保留 available_level 包含该 tier 的国家。""" + return [c for c in countries if tier in c.get('available_level', [])] + + +def pick_method(country, prefer=None): + """根据国家 methods 选择协议。返回 (protocol, method_config)。 + prefer 指定则优先；否则按 PROTOCOL_PRIORITY 回退。""" + methods = country.get('methods', {}) + if prefer and prefer in methods: + return prefer, methods[prefer] + for proto in PROTOCOL_PRIORITY: + if proto in methods: + return proto, methods[proto] + return None, None + + +def country_remark(country): + """clash/v2ray 备注：R.R.P.P. - <国家代码>（IPv6 加 "(IPv6)"）。""" + remark = f"R.R.P.P. - {country['country']}" + if country.get('ipv6_only'): + remark += ' (IPv6)' + return remark + + +# ---------- 各协议 → v2ray 分享链接 ---------- + +def vless_url(domain, port, cfg, remark): + params = { + 'security': 'reality', + 'encryption': 'none', + 'pbk': cfg['public_key'], + 'headerType': 'none', + 'fp': 'chrome', + 'type': 'tcp', + 'flow': cfg.get('flow', 'xtls-rprx-vision'), + 'sni': cfg['sni'], + 'sid': cfg['short_id'], + } + qs = '&'.join(f'{k}={v}' for k, v in params.items()) + return f"vless://{cfg['uuid']}@{domain}:{port}?{qs}#{urllib.parse.quote(remark, safe='')}" + + +def ss_url(domain, port, cfg, remark): + userinfo = base64.urlsafe_b64encode( + f"{cfg['method']}:{cfg['password']}".encode()).decode().rstrip('=') + return f"ss://{userinfo}@{domain}:{port}#{urllib.parse.quote(remark, safe='')}" + + +def http_url(domain, port, cfg, remark): + up = base64.urlsafe_b64encode( + f"{cfg['username']}:{cfg['password']}".encode()).decode().rstrip('=') + return f"http://{up}@{domain}:{port}#{urllib.parse.quote(remark, safe='')}" + + +URL_BUILDERS = { + 'vless': vless_url, + 'ss': ss_url, + 'http': http_url, +} + + +def countries_to_v2ray_sub(countries, prefer=None): + """生成 v2ray base64 订阅串。""" + lines = [] + for c in countries: + proto, cfg = pick_method(c, prefer) + if proto and proto in URL_BUILDERS: + lines.append(URL_BUILDERS[proto]( + c['domain'], cfg['port'], cfg, country_remark(c))) + return base64.b64encode('\n'.join(lines).encode('utf-8')).decode('ascii') + + +# ---------- 各协议 → Clash YAML ---------- + +def clash_proxy(domain, port, cfg, proto, name): + """节点 → Clash proxies 元素。""" + base = {'name': name, 'server': domain, 'port': port} + if proto == 'vless': + base.update({ + 'type': 'vless', + 'uuid': cfg['uuid'], + 'network': 'tcp', + 'tls': True, + 'udp': True, + 'flow': cfg.get('flow', 'xtls-rprx-vision'), + 'servername': cfg['sni'], + 'reality-opts': { + 'public-key': cfg['public_key'], + 'short-id': cfg['short_id'], + }, + 'client-fingerprint': 'chrome', + }) + elif proto == 'ss': + base.update({'type': 'ss', 'cipher': cfg['method'], 'password': cfg['password']}) + elif proto == 'http': + base.update({'type': 'http', 'username': cfg['username'], 'password': cfg['password']}) + return base + + +def countries_to_clash_yaml(countries, prefer=None): + """生成最小可用 Clash YAML。""" + proxies = [] + for c in countries: + proto, cfg = pick_method(c, prefer) + if proto: + proxies.append(clash_proxy( + c['domain'], cfg['port'], cfg, proto, country_remark(c))) + names = [p['name'] for p in proxies] + doc = { + 'proxies': proxies, + 'proxy-groups': [{ + 'name': 'RRPP', + 'type': 'select', + 'proxies': names, + }], + 'rules': ['MATCH,RRPP'], + } + return _dump_simple_yaml(doc) + + +def _dump_simple_yaml(obj, indent=0): + """极简 YAML 序列化（足够 proxies/groups/rules 结构）。""" + sp = ' ' * indent + lines = [] + if isinstance(obj, dict): + for k, v in obj.items(): + if isinstance(v, (dict, list)): + lines.append(f'{sp}{k}:') + lines.append(_dump_simple_yaml(v, indent + 1)) + else: + lines.append(f'{sp}{k}: {_yaml_scalar(v)}') + elif isinstance(obj, list): + for item in obj: + if isinstance(item, dict): + first = True + for k, v in item.items(): + prefix = f'{sp}- ' if first else f'{sp} ' + if isinstance(v, (dict, list)): + lines.append(f'{prefix}{k}:') + lines.append(_dump_simple_yaml(v, indent + 2)) + else: + lines.append(f'{prefix}{k}: {_yaml_scalar(v)}') + first = False + else: + lines.append(f'{sp}- {_yaml_scalar(item)}') + return '\n'.join(l for l in lines if l is not None) + + +def _yaml_scalar(v): + if isinstance(v, bool): + return 'true' if v else 'false' + if v is None: + return 'null' + if isinstance(v, (int, float)): + return str(v) + s = str(v) + if not s or any(c in s for c in ':#{}[],&*?|-<>=!%@`"\' \n') or s != s.strip(): + return json.dumps(s, ensure_ascii=False) + return s + + +# ---------- rrpp_client：旧格式（仅 http） ---------- + +def countries_to_rrpp_client(countries): + """返回旧格式节点列表（仅 http 协议），保证 RRPP 自建客户端兼容性。 + 格式：domain/port/country/username/password/remark（remark = provider + traffic）""" + out = [] + for c in countries: + m = c.get('methods', {}).get('http') + if not m: + continue + remark = f"{c.get('provider', '')} {c.get('traffic', '')}".strip() + if c.get('ipv6_only'): + remark += ' (IPv6)' + out.append({ + 'domain': c['domain'], + 'port': m['port'], + 'country': c['country'], + 'username': m['username'], + 'password': m['password'], + 'remark': remark, + }) + return out + + +# ---------- method 分发 ---------- + +def _make_rrpp_client(countries, prefer=None): + return json.dumps(countries_to_rrpp_client(countries), ensure_ascii=False, indent=2).encode('utf-8') + + +def _make_clash(countries, prefer=None): + return countries_to_clash_yaml(countries, prefer).encode('utf-8') + + +def _make_v2ray(countries, prefer=None): + return countries_to_v2ray_sub(countries, prefer).encode('utf-8') + + +METHOD_HANDLERS = { + 'rrpp_client': ('application/json', _make_rrpp_client), + 'clash': ('text/yaml', _make_clash), + 'v2ray': ('text/plain; charset=utf-8', _make_v2ray), +} + + +@app.route('/v1/rrpp/nodes_update') # 旧端点合并：等价于 method=rrpp_client +@app.route('/v1/rrpp/subscription') +def subscription(): + method = request.args.get('method', '') + if request.path.rstrip('/').endswith('/nodes_update'): + method = 'rrpp_client' + user = b64url_decode(request.args.get('user', '')) + password = b64url_decode(request.args.get('password', '')) + prefer = request.args.get('prefer', '').strip().lower() or None + + if not user or not password: + return '认证失败', 401 + + ok, err, tier = fetch_user_tier( + user, password, + client_ip=request.headers.get('X-Real-IP') or request.remote_addr or '') + if not ok: + return err, (502 if 'SSO' in err else 401) + if tier < 1: + return 'RRPP 服务未开通或未激活，请联系管理员放行', 403 + + handler = METHOD_HANDLERS.get(method) + if not handler: + return f'未知的订阅类型: {method}（可用: clash / v2ray / rrpp_client）', 404 + + try: + countries = load_countries() + except Exception: + return '节点配置加载失败，请联系管理员', 503 + + countries = filter_countries_by_tier(countries, tier) + if not countries: + return f'LV{tier} 暂无可用节点，请联系管理员', 503 + + mimetype, build = handler + return Response(build(countries, prefer), mimetype=mimetype) + + +if __name__ == '__main__': + # 仅本机监听，由 bj nginx 反代 + app.run(host='127.0.0.1', port=5010)