From 2ab16a68b71ae6eec8022d14cd48dd2a8308b005 Mon Sep 17 00:00:00 2001 From: Jonas Rabenstein Date: Thu, 13 Aug 2026 03:23:00 +0200 Subject: [PATCH] merge --- src/auth.rs | 281 +++++++++++++++++++++++++++++++----------------- src/db.rs | 117 +++++++++++++++----- src/main.rs | 300 +++++++++++++++++++++++++++++++++++++++++++++++----- src/web.rs | 9 +- 4 files changed, 556 insertions(+), 151 deletions(-) diff --git a/src/auth.rs b/src/auth.rs index ba487c1..bd9077a 100644 --- a/src/auth.rs +++ b/src/auth.rs @@ -1,6 +1,6 @@ use axum::{ - extract::{Query, State}, - response::{IntoResponse, Redirect}, + extract::{Form, Query, State}, + response::{Html, IntoResponse, Redirect}, }; use axum_extra::extract::cookie::{Cookie, CookieJar, SameSite}; use rand::{ @@ -8,6 +8,7 @@ use rand::{ rng, }; use serde::Deserialize; +use serde_json::Value; use crate::{AppState, db, intervals::IntervalsClient}; @@ -18,19 +19,19 @@ pub struct OAuthCallback { pub error: Option, } +#[derive(Debug, Deserialize)] +pub struct ApiKeyLoginForm { + pub api_key: String, +} + pub async fn start(State(state): State) -> impl IntoResponse { - // Important: ThreadRng must not live across an await. let oauth_state = { let mut rng = rng(); Alphanumeric.sample_string(&mut rng, 48) }; if let Err(error) = db::delete_old_oauth_states(&state.db).await { - tracing::error!( - %error, - "cannot clean OAuth states" - ); - + tracing::error!(%error, "cannot clean OAuth states"); return "internal error".into_response(); } @@ -39,11 +40,7 @@ pub async fn start(State(state): State) -> impl IntoResponse { .execute(&state.db) .await { - tracing::error!( - %error, - "cannot create OAuth state" - ); - + tracing::error!(%error, "cannot create OAuth state"); return "internal error".into_response(); } @@ -56,7 +53,6 @@ pub async fn start(State(state): State) -> impl IntoResponse { %error, "cannot construct OAuth authorize URL" ); - return "cannot construct OAuth URL".into_response(); } }; @@ -64,6 +60,155 @@ pub async fn start(State(state): State) -> impl IntoResponse { Redirect::to(&authorize_url).into_response() } +/// Development/workaround login using a personal Intervals.icu API key. +/// +/// GET /api-key/start shows the form. +/// POST /api-key/start validates the key against athlete/0, creates/updates +/// the local athlete and creates the normal county-sprints session. +pub async fn api_key_start() -> Html { + Html( + r#" + + + + +Mit Intervals.icu API-Key anmelden + + + +← Landkreis-Sprints +
+

Mit Intervals.icu API-Key anmelden

+

Dein persönlicher Intervals.icu API-Key wird direkt gegen Intervals.icu geprüft. Der Besitzer des API-Keys wird als Benutzer verwendet.

+
+ + + +
+

Für diese lokale Entwicklungs-/Workaround-Variante wird der API-Key serverseitig für die spätere Aktivitätsverarbeitung gespeichert. Für Produktion sollten wir ihn verschlüsselt speichern.

+
+ +"#.to_string(), + ) +} + +pub async fn api_key_login( + State(state): State, + Form(form): Form, +) -> impl IntoResponse { + let api_key = form.api_key.trim().to_string(); + + if api_key.is_empty() { + return Html(api_key_error("API-Key darf nicht leer sein")).into_response(); + } + + let client = IntervalsClient::with_api_key(state.config.clone(), api_key.clone()); + + let owner = match client.owner().await { + Ok(owner) => owner, + Err(error) => { + tracing::warn!(error = %error, "Intervals.icu API-key login failed"); + return Html(api_key_error( + "API-Key konnte bei Intervals.icu nicht verifiziert werden", + )) + .into_response(); + } + }; + + let athlete_id = match athlete_id_from_value(&owner) { + Some(id) => id, + None => { + tracing::error!("Intervals.icu owner response has no athlete id"); + return Html(api_key_error( + "Intervals.icu hat für den API-Key keinen Athleten geliefert", + )) + .into_response(); + } + }; + + let display_name = owner + .get("name") + .or_else(|| owner.get("display_name")) + .and_then(Value::as_str) + .unwrap_or("Intervals.icu athlete"); + + let athlete_db_id = + match db::upsert_api_key_athlete(&state.db, &athlete_id, display_name, &api_key).await { + Ok(id) => id, + Err(error) => { + tracing::error!(%error, "cannot store API-key athlete"); + return Html(api_key_error("Benutzer konnte nicht gespeichert werden")) + .into_response(); + } + }; + + let session_token = { + let mut rng = rng(); + Alphanumeric.sample_string(&mut rng, 64) + }; + + if let Err(error) = db::create_session(&state.db, athlete_db_id, &session_token).await { + tracing::error!(%error, "cannot create API-key login session"); + return Html(api_key_error("Session konnte nicht erstellt werden")).into_response(); + } + + let cookie = Cookie::build(("county_session", session_token)) + .path("/") + .http_only(true) + .same_site(SameSite::Lax) + .secure(state.config.cookie_secure); + + let jar = CookieJar::new().add(cookie); + + (jar, Redirect::to("/")).into_response() +} + +fn athlete_id_from_value(value: &Value) -> Option { + value + .get("id") + .and_then(Value::as_str) + .map(str::to_string) + .or_else(|| { + value + .get("id") + .and_then(Value::as_i64) + .map(|id| id.to_string()) + }) +} + +fn api_key_error(message: &str) -> String { + format!( + r#" + + + + +API-Key Login + + + +← zurück +
+

Login fehlgeschlagen

+

{}

+

Erneut versuchen

+
+ +"#, + escape_html(message) + ) +} + pub async fn callback( State(state): State, Query(query): Query, @@ -83,9 +228,8 @@ pub async fn callback( let state_row = sqlx::query( r#" DELETE FROM oauth_states - WHERE - state = $1 - AND created_at > now() - interval '10 minutes' + WHERE state = $1 + AND created_at > now() - interval '10 minutes' RETURNING state "#, ) @@ -95,17 +239,9 @@ pub async fn callback( match state_row { Ok(Some(_)) => {} - - Ok(None) => { - return "invalid or expired OAuth state".into_response(); - } - + Ok(None) => return "invalid or expired OAuth state".into_response(), Err(error) => { - tracing::error!( - %error, - "OAuth state validation failed" - ); - + tracing::error!(%error, "OAuth state validation failed"); return "internal error".into_response(); } } @@ -114,13 +250,8 @@ pub async fn callback( let response = match client.exchange_code(&code).await { Ok(response) => response, - Err(error) => { - tracing::error!( - %error, - "OAuth token exchange failed" - ); - + tracing::error!(%error, "OAuth token exchange failed"); return "OAuth token exchange failed".into_response(); } }; @@ -128,64 +259,24 @@ pub async fn callback( let (access_token, scope, athlete_id, display_name) = match IntervalsClient::token_data(&response) { Ok(value) => value, - Err(error) => { - tracing::error!( - %error, - "invalid OAuth token response" - ); - + tracing::error!(%error, "invalid OAuth token response"); return "invalid OAuth response".into_response(); } }; - let row = sqlx::query( - r#" - INSERT INTO athletes ( - intervals_athlete_id, - display_name, - access_token, - scopes - ) - VALUES ($1, $2, $3, $4) - ON CONFLICT (intervals_athlete_id) - DO UPDATE SET - display_name = EXCLUDED.display_name, - access_token = EXCLUDED.access_token, - scopes = EXCLUDED.scopes, - updated_at = now() - RETURNING id - "#, + let athlete_db_id = match db::upsert_oauth_athlete( + &state.db, + &athlete_id, + &display_name, + &access_token, + &scope, ) - .bind(&athlete_id) - .bind(&display_name) - .bind(&access_token) - .bind(&scope) - .fetch_one(&state.db) - .await; - - use sqlx::Row; - - let athlete_db_id: i64 = match row { - Ok(row) => match row.try_get("id") { - Ok(id) => id, - - Err(error) => { - tracing::error!( - %error, - "invalid athlete row" - ); - - return "internal error".into_response(); - } - }, - + .await + { + Ok(id) => id, Err(error) => { - tracing::error!( - %error, - "cannot store athlete" - ); - + tracing::error!(%error, "cannot store OAuth athlete"); return "cannot store athlete".into_response(); } }; @@ -196,11 +287,7 @@ pub async fn callback( }; if let Err(error) = db::create_session(&state.db, athlete_db_id, &session_token).await { - tracing::error!( - %error, - "cannot create session" - ); - + tracing::error!(%error, "cannot create session"); return "cannot create session".into_response(); } @@ -217,15 +304,19 @@ pub async fn callback( pub async fn logout(State(state): State, jar: CookieJar) -> impl IntoResponse { if let Some(cookie) = jar.get("county_session") { - let hash = db::hash_token(cookie.value()); - - let _ = sqlx::query("DELETE FROM sessions WHERE token_hash = $1") - .bind(&hash) - .execute(&state.db) - .await; + let _ = db::delete_session(&state.db, cookie.value()).await; } let jar = jar.remove(Cookie::build("county_session").path("/")); (jar, Redirect::to("/")).into_response() } + +fn escape_html(value: &str) -> String { + value + .replace('&', "&") + .replace('<', "<") + .replace('>', ">") + .replace('"', """) + .replace('\'', "'") +} diff --git a/src/db.rs b/src/db.rs index b003f3e..e906cb6 100644 --- a/src/db.rs +++ b/src/db.rs @@ -8,9 +8,7 @@ use crate::model::DetectedCrossing; pub fn hash_token(token: &str) -> Vec { let mut hasher = Sha256::new(); - hasher.update(token.as_bytes()); - hasher.finalize().to_vec() } @@ -19,10 +17,7 @@ pub async fn create_session(db: &PgPool, athlete_id: i64, token: &str) -> Result sqlx::query( r#" - INSERT INTO sessions ( - token_hash, - athlete_id - ) + INSERT INTO sessions (token_hash, athlete_id) VALUES ($1, $2) "#, ) @@ -39,12 +34,9 @@ pub async fn athlete_for_session(db: &PgPool, token: &str) -> Result Result Result { + let row = sqlx::query( + r#" + INSERT INTO athletes ( + intervals_athlete_id, + display_name, + access_token, + scopes + ) + VALUES ($1, $2, $3, 'API_KEY') + ON CONFLICT (intervals_athlete_id) + DO UPDATE SET + display_name = EXCLUDED.display_name, + access_token = EXCLUDED.access_token, + scopes = EXCLUDED.scopes, + updated_at = now() + RETURNING id + "#, + ) + .bind(intervals_athlete_id) + .bind(display_name) + .bind(api_key) + .fetch_one(db) + .await?; + + use sqlx::Row; + Ok(row.try_get("id")?) +} + +/// Create or update an athlete authenticated through OAuth. +pub async fn upsert_oauth_athlete( + db: &PgPool, + intervals_athlete_id: &str, + display_name: &str, + access_token: &str, + scopes: &str, +) -> Result { + let row = sqlx::query( + r#" + INSERT INTO athletes ( + intervals_athlete_id, + display_name, + access_token, + scopes + ) + VALUES ($1, $2, $3, $4) + ON CONFLICT (intervals_athlete_id) + DO UPDATE SET + display_name = EXCLUDED.display_name, + access_token = EXCLUDED.access_token, + scopes = EXCLUDED.scopes, + updated_at = now() + RETURNING id + "#, + ) + .bind(intervals_athlete_id) + .bind(display_name) + .bind(access_token) + .bind(scopes) + .fetch_one(db) + .await?; + + use sqlx::Row; + Ok(row.try_get("id")?) +} + pub async fn ensure_activity( tx: &mut Transaction<'_, Postgres>, athlete_id: i64, @@ -217,17 +280,8 @@ pub async fn ensure_activity( activity_json, processed_at ) - VALUES ( - $1, - $2, - $3, - $4, - NULL - ) - ON CONFLICT ( - athlete_id, - intervals_activity_id - ) + VALUES ($1, $2, $3, $4, NULL) + ON CONFLICT (athlete_id, intervals_activity_id) DO UPDATE SET start_time = EXCLUDED.start_time, activity_json = EXCLUDED.activity_json, @@ -243,7 +297,6 @@ pub async fn ensure_activity( .await?; use sqlx::Row; - Ok(row.try_get("id")?) } @@ -285,10 +338,7 @@ pub async fn save_crossing( $1, $2, $3, - ST_SetSRID( - ST_MakePoint($4, $5), - 4326 - ), + ST_SetSRID(ST_MakePoint($4, $5), 4326), $6, $7, $8 @@ -326,3 +376,14 @@ pub async fn mark_activity_processed( Ok(()) } + +pub async fn delete_session(db: &PgPool, token: &str) -> Result<()> { + let hash = hash_token(token); + + sqlx::query("DELETE FROM sessions WHERE token_hash = $1") + .bind(&hash) + .execute(db) + .await?; + + Ok(()) +} diff --git a/src/main.rs b/src/main.rs index fdac108..e5308c8 100644 --- a/src/main.rs +++ b/src/main.rs @@ -12,12 +12,12 @@ use anyhow::{Context, Result, anyhow}; use axum::{ Router, extract::{Path, Query, State}, - response::Html, + response::{Html, Json}, routing::{get, post}, }; use chrono::{DateTime, Duration, Utc}; use serde::{Deserialize, Serialize}; -use serde_json::Value; +use serde_json::{Value, json}; use sqlx::PgPool; use tower_http::trace::TraceLayer; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; @@ -56,9 +56,7 @@ async fn main() -> Result<()> { let config = Config::from_env()?; tracing::info!( - database_url = %redact_database_url( - &config.database_url - ), + database_url = %redact_database_url(&config.database_url), "connecting to PostgreSQL" ); @@ -81,6 +79,10 @@ async fn main() -> Result<()> { .route("/health", get(health)) .route("/oauth/start", get(auth::start)) .route("/oauth/callback", get(auth::callback)) + .route( + "/api-key/start", + get(auth::api_key_start).post(auth::api_key_login), + ) .route("/logout", get(auth::logout)) .route("/webhooks/intervals", post(webhook::receive)) .route("/api/leaderboard", get(leaderboard::api)) @@ -94,6 +96,11 @@ async fn main() -> Result<()> { "/sync/{athlete_id}/import-visible", post(sync_import_visible), ) + .route("/activity/{activity_id}", get(web::activity)) + .route("/groups", get(web::groups).post(web::save_group)) + .route("/groups/delete", post(web::delete_group)) + .route("/dev/sync/{api_key}", get(dev_sync_one)) + .route("/dev/sync/{api_key}/{athlete_id}", get(dev_sync)) .layer(TraceLayer::new_for_http()) .with_state(state); @@ -105,7 +112,6 @@ async fn main() -> Result<()> { ); axum::serve(listener, app).await?; - Ok(()) } @@ -119,7 +125,6 @@ fn redact_database_url(url: &str) -> String { return format!("{prefix}@***"); } } - url.to_string() } @@ -310,7 +315,6 @@ async fn sync_athlete( imported: false, }); } - let mut html = String::from( r#" @@ -512,7 +516,6 @@ input[type="datetime-local"] { html.push_str(""); } - html.push_str( r#"

← Leaderboard @@ -744,19 +747,245 @@ fn url_query(value: &str) -> String { urlencoding::encode(value).into_owned() } +async fn dev_sync( + State(state): State, + Path((api_key, athlete_id)): Path<(String, String)>, +) -> Result, axum::http::StatusCode> { + match dev_sync_inner(state, api_key, athlete_id).await { + Ok(result) => Ok(Json(result)), + Err(error) => { + tracing::error!(error = %error, "development sync failed"); + Err(axum::http::StatusCode::INTERNAL_SERVER_ERROR) + } + } +} + +async fn dev_sync_one( + State(state): State, + Path(api_key): Path, +) -> Result, axum::http::StatusCode> { + match dev_sync_inner(state, api_key, "0".to_string()).await { + Ok(result) => Ok(Json(result)), + Err(error) => { + tracing::error!(error = %error, "development sync failed"); + Err(axum::http::StatusCode::INTERNAL_SERVER_ERROR) + } + } +} + +async fn dev_sync_inner(state: AppState, api_key: String, athlete_id: String) -> Result { + if api_key.trim().is_empty() { + return Err(anyhow!("API key is empty")); + } + + let athlete_id = if athlete_id.trim().is_empty() { + "0".to_string() + } else { + athlete_id + }; + + let client = IntervalsClient::with_api_key(state.config.clone(), api_key); + + let resolved_athlete_id = if athlete_id == "0" { + let owner = client.owner().await?; + + owner + .get("id") + .and_then(Value::as_str) + .map(str::to_string) + .or_else(|| { + owner + .get("id") + .and_then(Value::as_i64) + .map(|id| id.to_string()) + }) + .context("Intervals.icu owner response has no athlete id")? + } else { + athlete_id.clone() + }; + + let newest = Utc::now().date_naive(); + let oldest = newest - chrono::Duration::days(state.config.dev_sync_days); + let oldest_string = oldest.format("%Y-%m-%d").to_string(); + let newest_string = newest.format("%Y-%m-%d").to_string(); + + tracing::info!( + athlete_id = %resolved_athlete_id, + oldest = %oldest_string, + newest = %newest_string, + "starting development activity sync" + ); + + let activities = client + .activities(&athlete_id, &oldest_string, &newest_string) + .await?; + + let activities_array = activities + .as_array() + .context("Intervals.icu activities response is not an array")?; + + ensure_dev_athlete(&state, &resolved_athlete_id, &client).await?; + + let mut processed = 0usize; + let mut skipped = 0usize; + let mut failed = 0usize; + let mut results = Vec::new(); + + for activity in activities_array { + let Some(activity_id) = activity.get("id").and_then(Value::as_str) else { + skipped += 1; + results.push(json!({"status": "skipped", "reason": "activity has no id"})); + continue; + }; + + match process_activity_with_client( + state.clone(), + resolved_athlete_id.clone(), + activity_id.to_string(), + &client, + ) + .await + { + Ok(()) => { + processed += 1; + results.push(json!({"id": activity_id, "status": "processed"})); + } + Err(error) => { + failed += 1; + tracing::error!( + activity_id = %activity_id, + error = %error, + "development activity processing failed" + ); + results.push(json!({ + "id": activity_id, + "status": "failed", + "error": error.to_string() + })); + } + } + } + + tracing::info!( + athlete_id = %resolved_athlete_id, + found = activities_array.len(), + processed, + skipped, + failed, + "development activity sync finished" + ); + + Ok(json!({ + "athlete_id": resolved_athlete_id, + "oldest": oldest_string, + "newest": newest_string, + "found": activities_array.len(), + "processed": processed, + "skipped": skipped, + "failed": failed, + "activities": results + })) +} + +async fn ensure_dev_athlete( + state: &AppState, + intervals_athlete_id: &str, + client: &IntervalsClient, +) -> Result<()> { + let athlete = client.athlete(intervals_athlete_id).await?; + + let display_name = athlete + .get("name") + .or_else(|| athlete.get("display_name")) + .and_then(Value::as_str) + .unwrap_or("Intervals.icu athlete"); + + let exists = sqlx::query( + r#" + SELECT id FROM athletes WHERE intervals_athlete_id = $1 + "#, + ) + .bind(intervals_athlete_id) + .fetch_optional(&state.db) + .await?; + + if exists.is_some() { + return Ok(()); + } + + sqlx::query( + r#" + INSERT INTO athletes ( + intervals_athlete_id, + access_token, + display_name, + scopes + ) + VALUES ($1, $2, $3, $4) + "#, + ) + .bind(intervals_athlete_id) + .bind("") + .bind(display_name) + .bind("") + .execute(&state.db) + .await + .context("cannot create local dev athlete")?; + + Ok(()) +} + +/// Normal processing entry point used by the webhook path. +/// The OAuth access token is loaded from the local athlete row. pub async fn process_activity( state: AppState, intervals_athlete_id: String, activity_id: String, +) -> Result<()> { + let access_token: String = sqlx::query_scalar( + r#" + SELECT access_token + FROM athletes + WHERE intervals_athlete_id = $1 + "#, + ) + .bind(&intervals_athlete_id) + .fetch_optional(&state.db) + .await? + .ok_or_else(|| { + anyhow!( + "no local athlete for Intervals.icu athlete {}", + intervals_athlete_id + ) + })?; + + if access_token.trim().is_empty() { + return Err(anyhow!( + "local athlete {} has no access token", + intervals_athlete_id + )); + } + + let client = IntervalsClient::with_api_key(state.config.clone(), access_token); + + process_activity_with_client(state, intervals_athlete_id, activity_id, &client).await +} + +/// Same processing pipeline, but with an explicitly supplied Intervals client. +/// The development endpoint uses this so its personal API key is never +/// written into the database. +pub async fn process_activity_with_client( + state: AppState, + intervals_athlete_id: String, + activity_id: String, + client: &IntervalsClient, ) -> Result<()> { let athlete = sqlx::query( r#" - SELECT - id, - access_token - FROM athletes - WHERE intervals_athlete_id = $1 - "#, + SELECT id + FROM athletes + WHERE intervals_athlete_id = $1 + "#, ) .bind(&intervals_athlete_id) .fetch_optional(&state.db) @@ -773,25 +1002,23 @@ pub async fn process_activity( let athlete_db_id: i64 = athlete.try_get("id")?; - let access_token: String = athlete.try_get("access_token")?; - - let client = IntervalsClient::with_api_key(state.config.clone(), access_token); - let activity = client.activity(&activity_id).await?; - let start_time = IntervalsClient::activity_start(&activity)?; - let streams = client.streams(&activity_id).await?; - let points = parse_track(&streams, start_time)?; + tracing::debug!( + activity_id = %activity_id, + points = points.len(), + "parsed activity GPS points" + ); + if points.len() < 2 { tracing::info!( activity_id = %activity_id, points = points.len(), "activity has insufficient GPS data" ); - return Ok(()); } @@ -820,13 +1047,14 @@ pub async fn process_activity( from = crossing.from_county_id, to = crossing.to_county_id, track_index = crossing.track_index, + lat = crossing.lat, + lon = crossing.lon, "county crossing detected" ); } } db::mark_activity_processed(&mut tx, activity_db_id).await?; - tx.commit().await?; tracing::info!( @@ -846,7 +1074,6 @@ fn parse_track(streams: &Value, start_time: DateTime) -> Result) -> Result) -> ResultMit Intervals.icu verbinden"#); + html.push_str( + r#"

"#, + ); } html.push_str("");