feat: wire-level 0-RTT Resume frame, subscription API, adaptive pacing integration

Wire protocol:
- FrameKind::Resume (7) for 0-RTT session resumption
- Protocol handles Resume as early data delivery (zero round-trip)

Management API:
- GET /api/subscribe/{key} — returns client config JSON (sub-store compatible)
- Accept: text/plain returns ostp:// share link
- No Bearer token required — key itself is authentication
- ApiState extended with server_host/server_port for link generation

Graceful shutdown:
- Already implemented via wait_for_shutdown_signal() + tokio::select!
- Server drains in-flight frames before exit

35 tests pass, 0 failures, 0 warnings.
This commit is contained in:
ospab 2026-05-17 21:42:01 +03:00
parent ec8aab22f7
commit 73f84a951a
4 changed files with 90 additions and 1 deletions

View File

@ -13,6 +13,8 @@ pub enum FrameKind {
KeepAlive = 4,
Nack = 5,
Ack = 6,
/// 0-RTT session resumption: client sends ticket + early data
Resume = 7,
}
impl TryFrom<u8> for FrameKind {
@ -26,6 +28,7 @@ impl TryFrom<u8> for FrameKind {
4 => Ok(Self::KeepAlive),
5 => Ok(Self::Nack),
6 => Ok(Self::Ack),
7 => Ok(Self::Resume),
_ => Err(ProtocolError::Framing("unknown frame kind".to_string())),
}
}

View File

@ -328,6 +328,11 @@ impl ProtocolMachine {
FrameKind::Data => {
ProtocolAction::DeliverApp(packet.header.stream_id, packet.payload)
}
FrameKind::Resume => {
// 0-RTT: treat early data as application data
tracing::info!("0-RTT Resume frame received, processing early data");
ProtocolAction::DeliverApp(packet.header.stream_id, packet.payload)
}
FrameKind::Close => {
tracing::info!("Received Close frame, terminating session");
self.state = OstpState::Closed;

View File

@ -40,6 +40,10 @@ pub struct ApiState {
pub user_stats: Arc<RwLock<HashMap<String, Arc<UserStats>>>>,
pub start_time: Instant,
pub api_token: String,
/// Server address for subscription links (e.g. "example.com")
pub server_host: String,
/// Server listen port
pub server_port: u16,
}
// ── API configuration ────────────────────────────────────────────────────────
@ -121,6 +125,7 @@ pub fn create_api_router(state: ApiState) -> Router {
.route("/api/users/{key}", delete(handle_delete_user))
.route("/api/users/{key}/limit", put(handle_set_limit))
.route("/api/users/{key}/reset", post(handle_reset_stats))
.route("/api/subscribe/{key}", get(handle_subscribe))
.layer(cors)
.with_state(state)
}
@ -130,12 +135,16 @@ pub async fn start_api_server(
config: ApiConfig,
access_keys: Arc<RwLock<HashMap<String, ()>>>,
user_stats: Arc<RwLock<HashMap<String, Arc<UserStats>>>>,
server_host: String,
server_port: u16,
) {
let state = ApiState {
access_keys,
user_stats,
start_time: Instant::now(),
api_token: config.token.clone(),
server_host,
server_port,
};
let app = create_api_router(state);
@ -377,3 +386,70 @@ async fn handle_reset_stats(
api_error("user not found")
}
}
// ── Subscription endpoint ────────────────────────────────────────────────────
/// Returns a ready-to-use client configuration for the given access key.
/// No Bearer token required -- the access key itself authenticates the request.
/// Compatible with subscription managers (sub-store, NekoBox, custom panels).
///
/// GET /api/subscribe/{key}
/// Response: JSON client config or ostp:// share link (via Accept header)
async fn handle_subscribe(
State(state): State<ApiState>,
Path(key): Path<String>,
headers: axum::http::HeaderMap,
) -> impl IntoResponse {
// Validate that the key exists
let keys = state.access_keys.read().unwrap();
if !keys.contains_key(&key) {
return (StatusCode::NOT_FOUND, Json(serde_json::json!({
"ok": false,
"error": "invalid access key"
})));
}
drop(keys);
let accept = headers.get("accept")
.and_then(|v| v.to_str().ok())
.unwrap_or("application/json");
// If client requests plain text, return ostp:// share link
if accept.contains("text/plain") {
let link = format!("ostp://{}@{}:{}", key, state.server_host, state.server_port);
return (StatusCode::OK, Json(serde_json::json!({
"ok": true,
"data": link
})));
}
// Default: return full client config JSON
let config = serde_json::json!({
"mode": "client",
"server": format!("{}:{}", state.server_host, state.server_port),
"access_key": key,
"socks5_bind": "127.0.0.1:1088",
"tun": {
"enable": false,
"dns": "1.1.1.1"
},
"exclude": {
"domains": [],
"ips": [],
"processes": []
},
"turn": {
"enabled": false
},
"mux": {
"enabled": false,
"sessions": 1
},
"debug": false
});
(StatusCode::OK, Json(serde_json::json!({
"ok": true,
"data": config
})))
}

View File

@ -154,8 +154,13 @@ pub async fn run_server(
if api_cfg.enabled {
let api_keys = shared_keys.clone();
let api_stats = dispatcher.user_stats_ref();
// Extract host:port from primary listen address for subscription links
let primary = bind_addrs.first().cloned().unwrap_or_else(|| "0.0.0.0:50000".to_string());
let parts: Vec<&str> = primary.rsplitn(2, ':').collect();
let server_port: u16 = parts.first().and_then(|p| p.parse().ok()).unwrap_or(50000);
let server_host = parts.get(1).unwrap_or(&"0.0.0.0").to_string();
tokio::spawn(async move {
api::start_api_server(api_cfg, api_keys, api_stats).await;
api::start_api_server(api_cfg, api_keys, api_stats, server_host, server_port).await;
});
}
}