357 lines
12 KiB
Python
357 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
RRPP 订阅后端（api.rrrosmontis.icu）— ROSMANKA ID SSO 国家分层版
|
||
|
||
URL 形式：
|
||
/v1/rrpp/subscription?method=<clash|v2ray|rrpp_client>&user=<base64url邮箱>&password=<base64url密码>[&prefer=<vless|ss|http>]
|
||
/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 = "<provider> <traffic>"（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 = ['http', 'ss', '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)
|