| 283 | } |
| 284 | |
| 285 | async fn recv_code(listener: TcpListener, expected_state: String) -> Result<String, AuthError> { |
| 286 | let (code_tx, mut code_rx) = tokio::sync::mpsc::channel::<Result<(String, String), AuthError>>(1); |
| 287 | let (stream, _) = listener.accept().await?; |
| 288 | let stream = TokioIo::new(stream); // Wrapper to implement Hyper IO traits for Tokio types. |
| 289 | let host = listener.local_addr()?.to_string(); |
| 290 | tokio::spawn(async move { |
| 291 | if let Err(err) = http1::Builder::new() |
| 292 | .serve_connection(stream, PkceHttpService { |
| 293 | code_tx: std::sync::Arc::new(code_tx), |
| 294 | host, |
| 295 | }) |
| 296 | .await |
| 297 | { |
| 298 | error!(?err, "Error occurred serving the connection"); |
| 299 | } |
| 300 | }); |
| 301 | match code_rx.recv().await { |
| 302 | Some(Ok((code, state))) => { |
| 303 | debug!(code = "<redacted>", state, "Received code and state"); |
| 304 | if state != expected_state { |
| 305 | return Err(AuthError::OAuthStateMismatch { |
| 306 | actual: state, |
| 307 | expected: expected_state, |
| 308 | }); |
| 309 | } |
| 310 | // Give time for the user to be redirected to index.html. |
| 311 | tokio::time::sleep(Duration::from_millis(200)).await; |
| 312 | Ok(code) |
| 313 | }, |
| 314 | Some(Err(err)) => { |
| 315 | // Give time for the user to be redirected to index.html. |
| 316 | tokio::time::sleep(Duration::from_millis(200)).await; |
| 317 | Err(err) |
| 318 | }, |
| 319 | None => Err(AuthError::OAuthMissingCode), |
| 320 | } |
| 321 | } |
| 322 | } |
| 323 | |
| 324 | type CodeSender = std::sync::Arc<tokio::sync::mpsc::Sender<Result<(String, String), AuthError>>>; |