pharos_native/runtime/
architect_chatgpt_http.rs

1//! Thin installed-Architect adapters to existing session, CSRF, storage and OAuth services.
2
3use super::*;
4use pharos_runtime_architect::chatgpt::{ConnectionError, InstalledArchitect};
5
6#[cfg(not(target_os = "wasi"))]
7use pharos_runtime_architect::chatgpt::service;
8
9struct OwnerContext {
10    authority: InstalledArchitect,
11    store: pharos_runtime_architect::database::ArchitectStore,
12    owner: i64,
13    session_hash: String,
14}
15
16/// Derives authority solely from the existing authenticated runtime context and operator registration.
17fn owner_context(app: &AppRuntime, request: &http::HttpRequest) -> Result<OwnerContext, Vec<u8>> {
18    let owner =
19        active_session_user_id(app, request.cookie.as_deref()).map_err(database_error_response)?;
20    if owner <= 0 {
21        return Err(http_text_response(
22            401,
23            "Unauthorized",
24            "Sign in to Architect first",
25        ));
26    }
27    if !actor_has_route_policy(app, owner, "manage_llm_vendors").map_err(database_error_response)? {
28        return Err(http_text_response(
29            403,
30            "Forbidden",
31            "LLM settings capability denied",
32        ));
33    }
34    let authority = InstalledArchitect::from_runtime(app).map_err(connection_failure)?;
35    let store = architect_task_store_for_app(app)
36        .map_err(|_| connection_failure(ConnectionError::Storage))?;
37    let token = request
38        .cookie
39        .as_deref()
40        .and_then(|cookie| dynamic::dynamic_cookie_value(cookie, "pharos_session"))
41        .ok_or_else(|| http_text_response(401, "Unauthorized", "Sign in to Architect first"))?;
42    let session_hash = native_session_token_hash(app, &token)
43        .ok_or_else(|| http_text_response(401, "Unauthorized", "Session unavailable"))?;
44    let context = OwnerContext {
45        authority,
46        store,
47        owner,
48        session_hash,
49    };
50    require_original_session(&context)?;
51    Ok(context)
52}
53
54/// Rechecks the actual session adapter after network work instead of trusting a cached login.
55fn require_original_session(context: &OwnerContext) -> Result<(), Vec<u8>> {
56    let now = SystemTime::now()
57        .duration_since(UNIX_EPOCH)
58        .map_err(|_| connection_failure(ConnectionError::Storage))?
59        .as_secs();
60    let session = context
61        .store
62        .adapter()
63        .active_session(
64            context.store.config(),
65            context.store.schema(),
66            &context.session_hash,
67            now,
68        )
69        .map_err(database_error_response)?;
70    if session.is_none_or(|session| session.user_id != context.owner) {
71        return Err(http_text_response(
72            401,
73            "Unauthorized",
74            "Architect session expired; sign in again",
75        ));
76    }
77    Ok(())
78}
79
80/// Converts safe service failures without including provider payloads or callback parameters.
81pub(super) fn connection_failure(error: ConnectionError) -> Vec<u8> {
82    let status = match error {
83        ConnectionError::Forbidden => 403,
84        ConnectionError::Storage | ConnectionError::Vault => 503,
85        ConnectionError::Busy => 409,
86        _ => 422,
87    };
88    http_text_response(status, http_reason_phrase(status), &error.to_string())
89}
90
91/// Routes only the fixed personal connection actions; it cannot issue or provision a local session.
92pub(super) fn architect_chatgpt_response(
93    app: &AppRuntime,
94    request: &http::HttpRequest,
95    method: &str,
96    action: pharos_runtime_architect::ArchitectHttpAction,
97) -> Vec<u8> {
98    #[cfg(target_os = "wasi")]
99    {
100        let _ = (app, request, method, action);
101        connection_failure(ConnectionError::Forbidden)
102    }
103    #[cfg(not(target_os = "wasi"))]
104    {
105        use pharos_runtime_architect::ArchitectHttpAction;
106        let context = match owner_context(app, request) {
107            Ok(context) => context,
108            Err(response) => return response,
109        };
110        if matches!(
111            action,
112            ArchitectHttpAction::ChatGptStatus | ArchitectHttpAction::ChatGptTest
113        ) {
114            let testing = action == ArchitectHttpAction::ChatGptTest;
115            if testing
116                && (method != "POST"
117                    || !dynamic::csrf_request_valid(
118                        request,
119                        dynamic::request_body_text(request).as_deref(),
120                    ))
121            {
122                return http_text_response(403, "Forbidden", "Security verification failed");
123            }
124            if !testing && method != "GET" {
125                return http_text_response(
126                    405,
127                    "Method Not Allowed",
128                    "Use GET for connection status",
129                );
130            }
131            let runtime = match tokio::runtime::Builder::new_current_thread()
132                .enable_all()
133                .build()
134            {
135                Ok(runtime) => runtime,
136                Err(_) => return connection_failure(ConnectionError::Storage),
137            };
138            let result = runtime.block_on(async {
139                if testing {
140                    service::test_connection(&context.authority, &context.store, context.owner)
141                        .await
142                } else {
143                    service::status(&context.authority, &context.store, context.owner).await
144                }
145            });
146            if let Err(response) = require_original_session(&context) {
147                return response;
148            }
149            return match result {
150                Ok(status) => match serde_json::to_string(&status) {
151                    Ok(body) => http_response_insert_header(
152                        http_json_response(200, "OK", &body, method),
153                        "Cache-Control",
154                        "no-store",
155                    ),
156                    Err(_) => connection_failure(ConnectionError::Storage),
157                },
158                Err(error) => connection_failure(error),
159            };
160        }
161        if method != "POST"
162            || !dynamic::csrf_request_valid(request, dynamic::request_body_text(request).as_deref())
163        {
164            return http_text_response(403, "Forbidden", "Security verification failed");
165        }
166        if action == ArchitectHttpAction::ChatGptCancel {
167            return match service::cancel_pending(&context.authority, context.owner) {
168                Ok(()) => http_json_response(
169                    200,
170                    "OK",
171                    &serde_json::json!({"status":"cancelled"}).to_string(),
172                    method,
173                ),
174                Err(error) => connection_failure(error),
175            };
176        }
177        if action == ArchitectHttpAction::ChatGptDisconnect {
178            let forget = dynamic::request_form_value(request, "forget_registration")
179                .is_some_and(|value| value == "1");
180            return match service::disconnect(
181                &context.authority,
182                &context.store,
183                context.owner,
184                forget,
185            ) {
186                Ok(()) => http_json_response(
187                    200,
188                    "OK",
189                    &serde_json::json!({"status":"disconnected"}).to_string(),
190                    method,
191                ),
192                Err(error) => connection_failure(error),
193            };
194        }
195        if action != ArchitectHttpAction::ChatGptConnect {
196            return connection_failure(ConnectionError::Forbidden);
197        }
198        let Some((_, root_key)) = architect_vault_paths(app) else {
199            return connection_failure(ConnectionError::Vault);
200        };
201        if ensure_architect_root_key(&root_key).is_err() {
202            return connection_failure(ConnectionError::Vault);
203        }
204        let (listener, port) = match bind_callback_listener(0) {
205            Ok(bound) => bound,
206            Err(error) => return connection_failure(error),
207        };
208        let redirect = format!("http://127.0.0.1:{port}/auth/callback");
209        let return_url = match context.authority.settings_return_url() {
210            Ok(url) => url,
211            Err(error) => return connection_failure(error),
212        };
213        let runtime = match tokio::runtime::Builder::new_current_thread()
214            .enable_all()
215            .build()
216        {
217            Ok(runtime) => runtime,
218            Err(_) => return connection_failure(ConnectionError::Storage),
219        };
220        let pending = match runtime.block_on(service::start_connection(
221            &context.authority,
222            &context.store,
223            context.owner,
224            context.session_hash,
225            &redirect,
226        )) {
227            Ok(pending) => pending,
228            Err(error) => return connection_failure(error),
229        };
230        let destination = pending.authorization_url().to_owned();
231        if std::thread::Builder::new()
232            .name("architect-chatgpt-callback".to_owned())
233            .stack_size(512 * 1024)
234            .spawn(move || {
235                run_callback_listener(listener, port, pending, context.store, return_url)
236            })
237            .is_err()
238        {
239            return connection_failure(ConnectionError::Busy);
240        }
241        redirect_no_store(&destination)
242    }
243}
244
245/// Binds only numeric loopback before authorization starts; failure cannot stage a connection.
246#[cfg(not(target_os = "wasi"))]
247fn bind_callback_listener(port: u16) -> Result<(std::net::TcpListener, u16), ConnectionError> {
248    let listener = std::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, port))
249        .map_err(|_| ConnectionError::Busy)?;
250    listener
251        .set_nonblocking(true)
252        .map_err(|_| ConnectionError::Busy)?;
253    let port = listener
254        .local_addr()
255        .map_err(|_| ConnectionError::Busy)?
256        .port();
257    Ok((listener, port))
258}
259
260/// Sends a redirect without authored markup or cacheable credential-bearing headers.
261fn redirect_no_store(destination: &str) -> Vec<u8> {
262    let response = http_text_response(303, "See Other", "");
263    let response = http_response_insert_header(response, "Location", destination);
264    let response = http_response_insert_header(response, "Cache-Control", "no-store");
265    http_response_insert_header(response, "Referrer-Policy", "no-referrer")
266}
267
268/// Owns a short-lived loopback listener in the existing runtime; no app server or session is created.
269#[cfg(not(target_os = "wasi"))]
270fn run_callback_listener(
271    listener: std::net::TcpListener,
272    port: u16,
273    pending: service::PendingConnection,
274    store: pharos_runtime_architect::database::ArchitectStore,
275    return_url: String,
276) {
277    use std::io::Write;
278    let deadline = std::time::Instant::now() + Duration::from_secs(300);
279    let mut count = 0;
280    while std::time::Instant::now() < deadline && !pending.is_cancelled() && count < 64 {
281        let (mut socket, peer) = match listener.accept() {
282            Ok(connection) => connection,
283            Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
284                std::thread::sleep(Duration::from_millis(50));
285                continue;
286            }
287            Err(_) => return,
288        };
289        count += 1;
290        if !peer.ip().is_loopback() {
291            continue;
292        }
293        let _ = socket.set_write_timeout(Some(Duration::from_secs(3)));
294        let query = match read_callback_query(&mut socket, port) {
295            Some(query) => query,
296            None => {
297                let _ = socket.write_all(&http_text_response(
298                    400,
299                    "Bad Request",
300                    "Invalid OAuth callback",
301                ));
302                continue;
303            }
304        };
305        if !pending.accepts_callback(&query) {
306            let _ = socket.write_all(&http_text_response(
307                403,
308                "Forbidden",
309                "OAuth callback state did not match",
310            ));
311            continue;
312        }
313        let result = tokio::runtime::Builder::new_current_thread()
314            .enable_all()
315            .build()
316            .map_err(|_| ConnectionError::Storage)
317            .and_then(|runtime| runtime.block_on(pending.complete(&store, &query)));
318        let state = if result.is_ok() {
319            "connected"
320        } else {
321            "failed"
322        };
323        let _ = socket.write_all(&redirect_no_store(&format!("{return_url}?chatgpt={state}")));
324        return;
325    }
326    if !pending.is_cancelled() {
327        pending.expire();
328    }
329}
330
331/// Reads only a bounded body-free GET callback through the existing HTTP parser with an absolute IO deadline.
332#[cfg(not(target_os = "wasi"))]
333fn read_callback_query(socket: &mut std::net::TcpStream, port: u16) -> Option<String> {
334    use std::io::Read;
335    let deadline = std::time::Instant::now() + Duration::from_secs(5);
336    let mut bytes = Vec::new();
337    let mut buffer = [0_u8; 1024];
338    let mut scan_from = 0;
339    loop {
340        socket
341            .set_read_timeout(Some(
342                deadline.checked_duration_since(std::time::Instant::now())?,
343            ))
344            .ok()?;
345        let read = socket.read(&mut buffer).ok()?;
346        if read == 0 || bytes.len() + read > 16 * 1024 {
347            return None;
348        }
349        bytes.extend_from_slice(&buffer[..read]);
350        if let Some(offset) = bytes[scan_from..]
351            .windows(4)
352            .position(|window| window == b"\r\n\r\n")
353        {
354            if scan_from + offset + 4 != bytes.len() {
355                return None;
356            }
357            break;
358        }
359        scan_from = bytes.len().saturating_sub(3);
360    }
361    let parsed = http::parse_http_request(&bytes).ok()?;
362    if parsed.method != "GET"
363        || parsed.headers.iter().any(|(name, value)| {
364            name == "transfer-encoding" || (name == "content-length" && value != "0")
365        })
366        || parsed
367            .headers
368            .iter()
369            .filter(|(name, _)| name == "content-length")
370            .count()
371            > 1
372    {
373        return None;
374    }
375    let mut hosts = parsed.headers.iter().filter(|(name, _)| name == "host");
376    let host = hosts.next()?.1.trim();
377    if host != format!("127.0.0.1:{port}") || hosts.next().is_some() {
378        return None;
379    }
380    let (path, query) = parsed.target.split_once('?')?;
381    if path != "/auth/callback" || query.len() > 8192 {
382        return None;
383    }
384    Some(query.to_owned())
385}
386
387#[cfg(all(test, not(target_os = "wasi")))]
388mod callback_tests {
389    use super::*;
390    use std::io::Write;
391
392    /// Exercises real socket contention and verifies the supported listener is loopback only.
393    #[test]
394    fn listener_failure_is_bounded_before_authorization() {
395        let (listener, port) = bind_callback_listener(0).unwrap();
396        assert_eq!(
397            listener.local_addr().unwrap().ip(),
398            std::net::Ipv4Addr::LOCALHOST
399        );
400        assert!(matches!(
401            bind_callback_listener(port),
402            Err(ConnectionError::Busy)
403        ));
404        drop(listener);
405        assert!(bind_callback_listener(port).is_ok());
406    }
407
408    /// Rejects alternate hosts, duplicate headers, request bodies and non-callback paths on real sockets.
409    #[test]
410    fn callback_http_boundary_rejects_untrusted_requests() {
411        for (target, headers, body, accepted) in [
412            ("/auth/callback?state=public", "", "", true),
413            ("/other?state=public", "", "", false),
414            (
415                "/auth/callback?state=public",
416                "Host: attacker.example\r\n",
417                "",
418                false,
419            ),
420            (
421                "/auth/callback?state=public",
422                "Content-Length: 1\r\n",
423                "x",
424                false,
425            ),
426            (
427                "/auth/callback?state=public",
428                "Content-Length: 0\r\nContent-Length: 0\r\n",
429                "",
430                false,
431            ),
432            (
433                "/auth/callback?state=public",
434                "Transfer-Encoding: chunked\r\n",
435                "",
436                false,
437            ),
438        ] {
439            let listener = std::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0)).unwrap();
440            let port = listener.local_addr().unwrap().port();
441            let mut client =
442                std::net::TcpStream::connect((std::net::Ipv4Addr::LOCALHOST, port)).unwrap();
443            client
444                .write_all(
445                    format!(
446                        "GET {target} HTTP/1.1\r\nHost: 127.0.0.1:{port}\r\n{headers}\r\n{body}"
447                    )
448                    .as_bytes(),
449                )
450                .unwrap();
451            client.shutdown(std::net::Shutdown::Write).unwrap();
452            let (mut server, _) = listener.accept().unwrap();
453            assert_eq!(read_callback_query(&mut server, port).is_some(), accepted);
454        }
455    }
456}
pharos_native/runtime/architect_chatgpt_http.rs

pharos_native/runtime/
architect_chatgpt_http.rs
