diff --git a/config.example.toml b/config.example.toml index 04aba81f..b6c4b802 100644 --- a/config.example.toml +++ b/config.example.toml @@ -20,6 +20,8 @@ admin_user = "mitosis_admin" admin_password = "mitosis_admin" access_token_private_path = "private.pem" access_token_public_path = "public.pem" +# access_token_expires_in applies to user access tokens only. Worker token lifetimes are +# chosen by the Worker itself, via its own lifetime setting. access_token_expires_in = "7d" heartbeat_timeout = "600s" file_log = false @@ -29,7 +31,11 @@ file_log = false coordinator_addr = "http://127.0.0.1:5000" polling_interval = "3m" heartbeat_interval = "5m" -lifetime = "7d" +# lifetime controls the worker JWT token lifetime. +# Use a duration such as "7d", "1h", or "30m". If omitted, the token never expires. +# The Coordinator's access_token_expires_in does not apply to worker tokens. The lifetime of +# worker token is completely controlled by worker side's settings. +# lifetime is not set # credential_path is not set # user is not set # password is not set @@ -40,8 +46,9 @@ file_log = false # log_path is not set. It will use the default rolling log file path if file_log is set to true # - If shared_log is enabled and log_path is not set, it will use workers.log in cache directory # - If shared_log is disabled and log_path is not set, it will use {worker_uuid}.log in cache directory -# retain is not set, default to false [client] # user = "mitosis_admin" # password = "mitosis_admin" +# refresh is not set, default to false. When true, the client refreshes the current login token and invalidates previous tokens. + diff --git a/guide/src/client/sdk.md b/guide/src/client/sdk.md index 46947c3e..79288af3 100644 --- a/guide/src/client/sdk.md +++ b/guide/src/client/sdk.md @@ -35,7 +35,7 @@ async fn main() -> Result<(), Box> { let login_args = LoginArgs { username: Some("username".to_string()), password: Some("password".to_string()), - retain: true, + refresh: false, }; client.user_login(login_args).await?; @@ -43,6 +43,32 @@ async fn main() -> Result<(), Box> { } ``` +### Authentication + +Refresh the current login token and invalidate previous tokens: + +```rust,ignore +client.refresh_token().await?; +``` + +Revoke all login tokens for the current user: + +```rust,ignore +client.revoke().await?; +``` + +Log in with username and password while invalidating previously issued tokens. +When `refresh` is true, `MitoClient::user_login` invalidates previously issued tokens. + +```rust,ignore +let login_args = LoginArgs { + username: Some("username".to_string()), + password: Some("password".to_string()), + refresh: true, +}; +client.user_login(login_args).await?; +``` + ### Task Management #### Submitting Tasks @@ -356,7 +382,7 @@ let config = ClientConfig { credential_path: Some(RelativePathBuf::from("/path/to/credentials")), user: Some("api-user".to_string()), password: Some("api-password".to_string()), - retain: true, // Keep existing login state + refresh: false, // Do not refresh the login token on setup }; ``` diff --git a/guide/src/guide/client.md b/guide/src/guide/client.md index 01685217..4d6f6720 100644 --- a/guide/src/guide/client.md +++ b/guide/src/guide/client.md @@ -32,6 +32,11 @@ If a user has never logged in or if his/her session has expired, the Client will Alternatively, they can directly specify their username (`-u`) or password (`-p`) during execution. Once authenticated, the Client will retain their credentials in a file for future use. +By default, logging in retains the previous login state, so tokens used by other clients remain valid. +To invalidate previously issued tokens while logging in with username and password, use `login --refresh`. +To refresh the current authenticated session without entering username and password again, use `refresh`. +To invalidate all login tokens for the current user, use `revoke`. + We recommend using the interactive mode for most operations, as it provides a more user-friendly experience. It will prompt you something like this: ```txt @@ -55,16 +60,18 @@ Run a mitosis client Usage: mito client [OPTIONS] [COMMAND] Commands: - admin Admin operations, including shutdown the coordinator, chaning user password, etc - auth Authenticate current user - login Login with username and password - users Manage users, including changing password, querying the accessible groups etc - groups Manage groups, including creating a group, querying groups, etc - tasks Manage tasks, including submitting a task, querying tasks, etc - workers Manage workers, including querying workers, cancel workers, etc - cmd Run an external command - quit Quit the client's interactive mode [aliases: exit] - help Print this message or the help of the given subcommand(s) + admin Admin operations, including shutdown the coordinator, chaning user password, etc + auth Authenticate current user + refresh Refresh current login token and invalidate previous tokens + revoke Revoke all login tokens of current user + login Login with username and password + users Manage users, including changing password, querying the accessible groups etc + groups Manage groups, including creating a group, querying groups, etc + tasks Manage tasks, including submitting a task, querying tasks, etc + workers Manage workers, including querying workers, cancel workers, etc + cmd Run an external command + quit Quit the client's interactive mode [aliases: exit] + help Print this message or the help of the given subcommand(s) Options: --config @@ -79,8 +86,8 @@ Options: The password of the user -i, --interactive Enable interactive mode - --retain - Whether to retain the previous login state without refetching the credential + --refresh + Refresh current login token and invalidate previous tokens during client setup -h, --help Print help -V, --version @@ -109,6 +116,38 @@ For the rest of this section, we will explain the common use cases of the Client For the sake of convenience, we will assume that the user is already in interactive mode. And for the direct executing mode, it only requires adding "mito client" at the front. +## Authentication commands + +Input `auth` to show the current authenticated user: + +```txt +auth +``` + +Input `login` to log in with username and password. Login retains previous tokens by default. + +```txt +login +``` + +Input `login --refresh` to log in with username and password and invalidate previously issued tokens. + +```txt +login --refresh +``` + +Input `refresh` to refresh the current valid token and invalidate previous tokens. + +```txt +refresh +``` + +Input `revoke` to invalidate all login tokens for the current user. In interactive mode, the client exits after a successful `revoke`. + +```txt +revoke +``` + ## `admin` sub-commands Input `help admin` to show the help message of the `admin` sub-commands: diff --git a/guide/src/guide/coordinator.md b/guide/src/guide/coordinator.md index 14968a56..33ba9d81 100644 --- a/guide/src/guide/coordinator.md +++ b/guide/src/guide/coordinator.md @@ -79,6 +79,8 @@ admin_user = "mitosis_admin" admin_password = "mitosis_admin" access_token_private_path = "private.pem" access_token_public_path = "public.pem" +# access_token_expires_in applies to user access tokens only. Worker token lifetimes are +# chosen by the Worker itself, via its own lifetime setting. access_token_expires_in = "7d" heartbeat_timeout = "600s" file_log = false diff --git a/guide/src/guide/worker.md b/guide/src/guide/worker.md index e0bfbbff..dae920f3 100644 --- a/guide/src/guide/worker.md +++ b/guide/src/guide/worker.md @@ -11,7 +11,7 @@ Before starting a Worker, we need to understand the environment inside a Worker. The Worker will spawn a new process for each task it runs and set up the following environment variables: - `MITO_TASK_UUID`: This will be set to the UUID of the task being executed. -- `MITO_NEW_TASK`: This will be set to the path a file where you can write a new task specification (i.e., SubmitTaskReq) in json format for the Worker to submit it on behalf of you, as a downstream task of the current task. +- `MITO_NEW_TASK`: This will be set to the path to a file where you can write a new task specification (i.e., SubmitTaskReq) in json format for the Worker to submit it on behalf of you, as a downstream task of the current task. - `MITO_UPSTREAM_TASK_UUID`: This will be set to the UUID of the upstream task if the current task is submitted by another task while running. - `MITO_RESOURCE_DIR`: This will be set to the path of a directory where you can find the resources (i.e., attachments) of the task. - `MITO_RESULT_DIR`: This will be set to the path of a directory where you can store the results of the task. The Worker will pack the directory and upload it as the artifacts of the task if it is not empty. @@ -20,13 +20,13 @@ The Worker will spawn a new process for each task it runs and set up the followi ## Starting a Worker To start a Worker, you need to provide a TOML file that configures the Worker. -The TOML file specifies the Worker's configuration, such as the polling (fetching) interval, the URL of the Coordinator, and the the groups allowed to submit tasks to it. +The TOML file specifies the Worker's configuration, such as the polling (fetching) interval, the URL of the Coordinator, and the groups allowed to submit tasks to it. All configuration options are optional and have default values. The Worker will merge the configuration from the file and the command-line arguments according to the following order (the latter overrides the former): ```md -DEFAULT <- `$CONFIG_DIR`/mitosis/config.toml <- config file specified by `cli.config` or loal `config.toml` <- env prefixed by `MITO_` <- cli arguments +DEFAULT <- `$CONFIG_DIR`/mitosis/config.toml <- config file specified by `cli.config` or local `config.toml` <- env prefixed by `MITO_` <- cli arguments `$CONFIG_DIR` will be different on different platforms: @@ -42,7 +42,11 @@ Here is an example of a Worker configuration file (you can also refer to `config coordinator_addr = "http://127.0.0.1:5000" polling_interval = "3m" heartbeat_interval = "5m" -lifetime = "7d" +# lifetime controls the worker JWT token lifetime. +# Use a duration such as "7d", "1h", or "30m". +# If not set, the JWT is valid forever. +# The Coordinator's access_token_expires_in does not apply to worker tokens. +# lifetime is not set # credential_path is not set # user is not set # password is not set @@ -53,7 +57,6 @@ file_log = false # log_path is not set. It will use the default rolling log file path if file_log is set to true # - If shared_log is enabled and log_path is not set, it will use workers.log in cache directory # - If shared_log is disabled and log_path is not set, it will use {worker_uuid}.log in cache directory -# lifetime is not set, default to the coordinator's setting ``` To start a Worker, run the following command: @@ -120,7 +123,7 @@ Options: --file-log Enable logging to file --lifetime - The lifetime of the worker to alive (e.g., 7d, 1year) + The lifetime of the worker token (e.g., 7d, 1year). If not given, the worker token is valid forever -h, --help Print help -V, --version diff --git a/netmito/src/api/mod.rs b/netmito/src/api/mod.rs index d0335160..c26fba53 100644 --- a/netmito/src/api/mod.rs +++ b/netmito/src/api/mod.rs @@ -50,6 +50,20 @@ pub fn router(st: InfraPool, cancel_token: CancellationToken) -> Router { get(|| async { (StatusCode::OK, Json(json!({"status": "ok"}))) }), ) .route("/login", post(users::user_login)) + .route( + "/refresh", + post(users::refresh_token).layer(middleware::from_fn_with_state( + st.clone(), + user_auth_middleware, + )), + ) + .route( + "/revoke", + post(users::revoke).layer(middleware::from_fn_with_state( + st.clone(), + user_auth_middleware, + )), + ) .route( "/redis", get(query_redis_connection_info).layer(middleware::from_fn_with_state( diff --git a/netmito/src/api/users.rs b/netmito/src/api/users.rs index 448f76cb..864bf391 100644 --- a/netmito/src/api/users.rs +++ b/netmito/src/api/users.rs @@ -33,17 +33,22 @@ pub async fn user_login( State(pool): State, Json(req): Json, ) -> Result, ApiError> { - let token = - service::auth::user_login(&pool.db, &req.username, &req.md5_password, req.retain, addr) - .await - .map_err(|e| match e { - crate::error::Error::AuthError(err) => ApiError::AuthError(err), - crate::error::Error::ApiError(e) => e, - _ => { - tracing::error!("{}", e); - ApiError::InternalServerError - } - })?; + let token = service::auth::user_login( + &pool.db, + &req.username, + &req.md5_password, + req.refresh, + addr, + ) + .await + .map_err(|e| match e { + crate::error::Error::AuthError(err) => ApiError::AuthError(err), + crate::error::Error::ApiError(e) => e, + _ => { + tracing::error!("{}", e); + ApiError::InternalServerError + } + })?; Ok(Json(UserLoginResp { token })) } @@ -51,6 +56,43 @@ pub async fn user_auth(Extension(u): Extension) -> String { u.username } +pub async fn refresh_token( + ConnectInfo(addr): ConnectInfo, + Extension(u): Extension, + State(pool): State, +) -> Result, ApiError> { + let token = service::auth::refresh_user_token(&pool.db, u.id, addr) + .await + .map_err(|e| match e { + crate::error::Error::AuthError(err) => ApiError::AuthError(err), + crate::error::Error::ApiError(e) => e, + _ => { + tracing::error!("{}", e); + ApiError::InternalServerError + } + })?; + + Ok(Json(UserLoginResp { token })) +} + +pub async fn revoke( + Extension(u): Extension, + State(pool): State, +) -> Result<(), ApiError> { + service::auth::revoke(&pool.db, u.id) + .await + .map_err(|e| match e { + crate::error::Error::AuthError(err) => ApiError::AuthError(err), + crate::error::Error::ApiError(e) => e, + _ => { + tracing::error!("{}", e); + ApiError::InternalServerError + } + })?; + + Ok(()) +} + pub async fn change_password( ConnectInfo(addr): ConnectInfo, Extension(u): Extension, diff --git a/netmito/src/client/http.rs b/netmito/src/client/http.rs index c62bff3a..96277804 100644 --- a/netmito/src/client/http.rs +++ b/netmito/src/client/http.rs @@ -1,5 +1,4 @@ use std::collections::HashMap; -use std::path::PathBuf; use figment::value::magic::RelativePathBuf; use reqwest::Client; @@ -8,61 +7,59 @@ use uuid::Uuid; use crate::{ entity::content::ArtifactContentType, - error::{get_error_from_resp, map_reqwest_err, RequestError}, + error::{get_error_from_resp, map_reqwest_err, CredentialGuardError, RequestError}, schema::*, - service::auth::cred::{get_user_credential, modify_or_append_credential}, + service::auth::{ + cred::get_user_credential, credential_guard::CredentialGuard, get_and_prompt_username, + }, }; pub struct MitoHttpClient { http_client: Client, url: Url, - credential: String, - credential_path: PathBuf, + credential_guard: CredentialGuard, } impl MitoHttpClient { - pub fn new(mut coordinator_addr: Url) -> Self { + pub async fn new( + mut coordinator_addr: Url, + credential_path: Option, + ) -> crate::error::Result { let http_client = Client::new(); + let credential_guard = CredentialGuard::new( + credential_path.map(|credential_path| credential_path.relative()), + &coordinator_addr, + ) + .await; coordinator_addr.set_path("/"); - Self { + + Ok(Self { http_client, url: coordinator_addr, - credential: String::new(), - credential_path: PathBuf::new(), - } + credential_guard, + }) } pub async fn connect( &mut self, - credential_path: Option, user: Option, password: Option, - retain: bool, + refresh: bool, ) -> crate::error::Result { - let client_credential_path = credential_path - .as_ref() - .map(|p| p.relative()) - .or_else(|| { - dirs::config_dir().map(|mut p| { - p.push("mitosis"); - p.push("credentials"); - p - }) - }) - .ok_or(crate::error::Error::ConfigError(Box::new( - figment::Error::from("credential path not found"), - )))?; - let (username, credential) = get_user_credential( - credential_path.as_ref(), + let username = match user { + Some(name) => name, + None => get_and_prompt_username(None, "Please input username")?, + }; + get_user_credential( + &mut self.credential_guard, &self.http_client, self.url.clone(), - user, + username.clone(), password, - retain, + refresh, ) .await?; - self.credential_path = client_credential_path; - self.credential = credential; + Ok(username) } @@ -83,11 +80,16 @@ impl MitoHttpClient { } pub async fn get_redis_connection_info(&mut self) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path("redis"); let resp = self .http_client .get(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .send() .await .map_err(map_reqwest_err)?; @@ -103,11 +105,16 @@ impl MitoHttpClient { } pub async fn user_auth(&mut self) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path("auth"); let resp = self .http_client .get(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .send() .await .map_err(map_reqwest_err)?; @@ -128,19 +135,71 @@ impl MitoHttpClient { .send() .await .map_err(map_reqwest_err)?; + self.credential_guard + .load_credential(req.username.clone()) + .await; if resp.status().is_success() { let resp = resp .json::() .await .map_err(RequestError::from)?; - self.credential = resp.token; - if self.credential_path.exists() { - if let Some(parent) = self.credential_path.parent() { - tokio::fs::create_dir_all(parent).await?; - } - modify_or_append_credential(&self.credential_path, &req.username, &self.credential) - .await?; - } + self.credential_guard + .save_credential(None, &resp.token) + .await; + Ok(()) + } else { + Err(get_error_from_resp(resp).await.into()) + } + } + + pub async fn refresh_token(&mut self) -> crate::error::Result<()> { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; + self.url.set_path("refresh"); + let resp = self + .http_client + .post(self.url.as_str()) + .bearer_auth(credential) + .send() + .await + .map_err(map_reqwest_err)?; + + if resp.status().is_success() { + let resp = resp + .json::() + .await + .map_err(RequestError::from)?; + + self.credential_guard + .save_credential(None, &resp.token) + .await; + + Ok(()) + } else { + Err(get_error_from_resp(resp).await.into()) + } + } + + pub async fn revoke(&mut self) -> crate::error::Result<()> { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; + self.url.set_path("revoke"); + let resp = self + .http_client + .post(self.url.as_str()) + .bearer_auth(credential) + .send() + .await + .map_err(map_reqwest_err)?; + + if resp.status().is_success() { + self.credential_guard.remove_credential(None).await; Ok(()) } else { Err(get_error_from_resp(resp).await.into()) @@ -152,12 +211,18 @@ impl MitoHttpClient { username: String, req: AdminChangePasswordReq, ) -> crate::error::Result<()> { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; + self.url .set_path(&format!("admin/users/{username}/password")); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -174,11 +239,16 @@ impl MitoHttpClient { username: String, req: UserChangePasswordReq, ) -> crate::error::Result<()> { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path(&format!("users/{username}/password")); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -188,14 +258,9 @@ impl MitoHttpClient { .json::() .await .map_err(RequestError::from)?; - self.credential = resp.token; - if self.credential_path.exists() { - if let Some(parent) = self.credential_path.parent() { - tokio::fs::create_dir_all(parent).await?; - } - modify_or_append_credential(&self.credential_path, &username, &self.credential) - .await?; - } + self.credential_guard + .save_credential(None, &resp.token) + .await; Ok(()) } else { Err(get_error_from_resp(resp).await.into()) @@ -203,11 +268,16 @@ impl MitoHttpClient { } pub async fn admin_create_user(&mut self, req: CreateUserReq) -> crate::error::Result<()> { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path("admin/users"); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -220,11 +290,16 @@ impl MitoHttpClient { } pub async fn admin_delete_user(&mut self, username: String) -> crate::error::Result<()> { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path(&format!("admin/users/{username}")); let resp = self .http_client .delete(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .send() .await .map_err(map_reqwest_err)?; @@ -240,6 +315,11 @@ impl MitoHttpClient { uuid: Uuid, force: bool, ) -> crate::error::Result<()> { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path(&format!("admin/workers/{uuid}")); if force { self.url.set_query(Some("op=force")) @@ -247,7 +327,7 @@ impl MitoHttpClient { let resp = self .http_client .delete(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .send() .await .map_err(map_reqwest_err)?; @@ -259,11 +339,16 @@ impl MitoHttpClient { } pub async fn user_create_group(&mut self, req: CreateGroupReq) -> crate::error::Result<()> { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path("groups"); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -276,11 +361,16 @@ impl MitoHttpClient { } pub async fn get_task_by_uuid(&mut self, uuid: Uuid) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path(&format!("tasks/{uuid}")); let resp = self .http_client .get(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .send() .await .map_err(map_reqwest_err)?; @@ -300,6 +390,11 @@ impl MitoHttpClient { uuid: Uuid, content_type: ArtifactContentType, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; let content_serde_val = serde_json::to_value(content_type)?; let content_serde_str = content_serde_val.as_str().unwrap_or("result"); self.url.set_path(&format!( @@ -308,7 +403,7 @@ impl MitoHttpClient { let resp = self .http_client .get(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .send() .await .map_err(map_reqwest_err)?; @@ -328,12 +423,17 @@ impl MitoHttpClient { group_name: &str, key: &str, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url .set_path(&format!("groups/{group_name}/download/attachments/{key}")); let resp = self .http_client .get(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .send() .await .map_err(map_reqwest_err)?; @@ -396,11 +496,16 @@ impl MitoHttpClient { &mut self, req: ArtifactsDownloadByFilterReq, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path("tasks/download/artifacts"); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -420,11 +525,16 @@ impl MitoHttpClient { &mut self, req: ArtifactsDownloadByUuidsReq, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path("tasks/download/artifacts/list"); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -445,12 +555,17 @@ impl MitoHttpClient { group_name: &str, req: AttachmentsDownloadByFilterReq, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url .set_path(&format!("groups/{group_name}/download/attachments")); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -471,12 +586,17 @@ impl MitoHttpClient { group_name: &str, req: AttachmentsDownloadByKeysReq, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url .set_path(&format!("groups/{group_name}/download/attachments/list")); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -498,6 +618,11 @@ impl MitoHttpClient { content_type: ArtifactContentType, admin: bool, ) -> crate::error::Result<()> { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; let content_serde_val = serde_json::to_value(content_type)?; let content_serde_str = content_serde_val.as_str().unwrap_or("result"); if admin { @@ -510,7 +635,7 @@ impl MitoHttpClient { let resp = self .http_client .delete(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .send() .await .map_err(map_reqwest_err)?; @@ -527,6 +652,11 @@ impl MitoHttpClient { key: &str, admin: bool, ) -> crate::error::Result<()> { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; if admin { self.url .set_path(&format!("admin/groups/{group_name}/attachments/{key}")); @@ -537,7 +667,7 @@ impl MitoHttpClient { let resp = self .http_client .delete(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .send() .await .map_err(map_reqwest_err)?; @@ -552,12 +682,17 @@ impl MitoHttpClient { group_name: &str, key: &str, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url .set_path(&format!("groups/{group_name}/attachments/{key}")); let resp = self .http_client .get(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .send() .await .map_err(map_reqwest_err)?; @@ -577,12 +712,17 @@ impl MitoHttpClient { username: &str, req: ChangeUserGroupQuota, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url .set_path(&format!("admin/users/{username}/group-quota")); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -603,12 +743,17 @@ impl MitoHttpClient { group_name: &str, req: ChangeGroupStorageQuotaReq, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url .set_path(&format!("admin/groups/{group_name}/storage-quota")); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -628,11 +773,16 @@ impl MitoHttpClient { &mut self, req: TasksQueryReq, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path("tasks/query"); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -653,12 +803,17 @@ impl MitoHttpClient { group_name: &str, req: AttachmentsQueryReq, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url .set_path(&format!("groups/{group_name}/attachments/query")); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -678,11 +833,16 @@ impl MitoHttpClient { &mut self, uuid: Uuid, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path(&format!("workers/{uuid}")); let resp = self .http_client .get(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .send() .await .map_err(map_reqwest_err)?; @@ -701,11 +861,16 @@ impl MitoHttpClient { &mut self, req: WorkersQueryReq, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path("workers/query"); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -725,11 +890,16 @@ impl MitoHttpClient { &mut self, group_name: &str, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path(&format!("groups/{group_name}")); let resp = self .http_client .get(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .send() .await .map_err(map_reqwest_err)?; @@ -745,11 +915,16 @@ impl MitoHttpClient { } pub async fn get_user_groups_roles(&mut self) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path("users/groups"); let resp = self .http_client .get(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .send() .await .map_err(map_reqwest_err)?; @@ -768,11 +943,16 @@ impl MitoHttpClient { &mut self, req: SubmitTaskReq, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path("tasks"); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -804,11 +984,16 @@ impl MitoHttpClient { uuid: Uuid, req: UploadArtifactReq, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path(&format!("tasks/{uuid}/artifacts")); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -829,12 +1014,17 @@ impl MitoHttpClient { group_name: &str, req: UploadAttachmentReq, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url .set_path(&format!("groups/{group_name}/attachments")); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -855,6 +1045,11 @@ impl MitoHttpClient { uuid: Uuid, force: bool, ) -> crate::error::Result<()> { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path(&format!("workers/{uuid}")); if force { self.url.set_query(Some("op=force")) @@ -862,7 +1057,7 @@ impl MitoHttpClient { let resp = self .http_client .delete(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .send() .await .map_err(map_reqwest_err)?; @@ -877,11 +1072,16 @@ impl MitoHttpClient { &mut self, req: WorkersShutdownByFilterReq, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path("workers/shutdown"); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -901,11 +1101,16 @@ impl MitoHttpClient { &mut self, req: WorkersShutdownByUuidsReq, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path("workers/shutdown/list"); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -926,11 +1131,16 @@ impl MitoHttpClient { uuid: Uuid, req: ReplaceWorkerTagsReq, ) -> crate::error::Result<()> { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path(&format!("workers/{uuid}/tags")); let resp = self .http_client .put(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -947,11 +1157,16 @@ impl MitoHttpClient { uuid: Uuid, req: ReplaceWorkerLabelsReq, ) -> crate::error::Result<()> { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path(&format!("workers/{uuid}/labels")); let resp = self .http_client .put(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -968,11 +1183,16 @@ impl MitoHttpClient { uuid: Uuid, req: UpdateGroupWorkerRoleReq, ) -> crate::error::Result<()> { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path(&format!("workers/{uuid}/groups")); let resp = self .http_client .put(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -989,6 +1209,11 @@ impl MitoHttpClient { uuid: Uuid, req: RemoveGroupWorkerRoleReq, ) -> crate::error::Result<()> { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path(&format!("workers/{uuid}/groups")); let params = req .groups @@ -998,7 +1223,7 @@ impl MitoHttpClient { let resp = self .http_client .delete(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .query(¶ms) .send() .await @@ -1011,11 +1236,16 @@ impl MitoHttpClient { } pub async fn cancel_task_by_uuid(&mut self, uuid: Uuid) -> crate::error::Result<()> { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path(&format!("tasks/{uuid}")); let resp = self .http_client .delete(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .send() .await .map_err(map_reqwest_err)?; @@ -1030,11 +1260,16 @@ impl MitoHttpClient { &mut self, req: TasksCancelByFilterReq, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path("tasks/cancel"); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -1054,11 +1289,16 @@ impl MitoHttpClient { &mut self, req: TasksCancelByUuidsReq, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path("tasks/cancel/list"); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -1079,11 +1319,16 @@ impl MitoHttpClient { uuid: Uuid, req: UpdateTaskLabelsReq, ) -> crate::error::Result<()> { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path(&format!("tasks/{uuid}/labels")); let resp = self .http_client .put(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -1100,11 +1345,16 @@ impl MitoHttpClient { uuid: Uuid, req: ChangeTaskReq, ) -> crate::error::Result<()> { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path(&format!("tasks/{uuid}")); let resp = self .http_client .put(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -1121,11 +1371,16 @@ impl MitoHttpClient { group_name: &str, req: UpdateUserGroupRoleReq, ) -> crate::error::Result<()> { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path(&format!("groups/{group_name}/users")); let resp = self .http_client .put(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -1142,6 +1397,11 @@ impl MitoHttpClient { group_name: &str, req: RemoveUserGroupRoleReq, ) -> crate::error::Result<()> { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path(&format!("groups/{group_name}/users")); let params = req .users @@ -1151,7 +1411,7 @@ impl MitoHttpClient { let resp = self .http_client .delete(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .query(¶ms) .send() .await @@ -1167,11 +1427,16 @@ impl MitoHttpClient { &mut self, req: ArtifactsDeleteByFilterReq, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path("tasks/delete/artifacts"); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -1191,11 +1456,16 @@ impl MitoHttpClient { &mut self, req: ArtifactsDeleteByUuidsReq, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path("tasks/delete/artifacts/list"); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -1216,12 +1486,17 @@ impl MitoHttpClient { group_name: &str, req: AttachmentsDeleteByFilterReq, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url .set_path(&format!("groups/{group_name}/delete/attachments")); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -1242,12 +1517,17 @@ impl MitoHttpClient { group_name: &str, req: AttachmentsDeleteByKeysReq, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url .set_path(&format!("groups/{group_name}/delete/attachments/list")); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -1267,11 +1547,16 @@ impl MitoHttpClient { &mut self, req: TasksSubmitReq, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path("tasks/submit"); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -1291,11 +1576,16 @@ impl MitoHttpClient { &mut self, req: ShutdownReq, ) -> crate::error::Result<()> { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path("admin/shutdown"); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -1311,11 +1601,16 @@ impl MitoHttpClient { &mut self, req: CreateTaskSuiteReq, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path("suites"); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -1335,11 +1630,16 @@ impl MitoHttpClient { &mut self, req: TaskSuitesQueryReq, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path("suites/query"); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&req) .send() .await @@ -1356,11 +1656,16 @@ impl MitoHttpClient { } pub async fn get_task_suite(&mut self, uuid: Uuid) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path(&format!("suites/{uuid}")); let resp = self .http_client .get(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .send() .await .map_err(map_reqwest_err)?; @@ -1376,11 +1681,16 @@ impl MitoHttpClient { } pub async fn close_task_suite(&mut self, uuid: Uuid) -> crate::error::Result<()> { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path(&format!("suites/{uuid}/close")); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .send() .await .map_err(map_reqwest_err)?; @@ -1392,6 +1702,11 @@ impl MitoHttpClient { } pub async fn cancel_task_suite(&mut self, uuid: Uuid, force: bool) -> crate::error::Result<()> { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url.set_path(&format!("suites/{uuid}")); if force { self.url.set_query(Some("op=force")); @@ -1399,7 +1714,7 @@ impl MitoHttpClient { let resp = self .http_client .delete(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .send() .await .map_err(map_reqwest_err)?; @@ -1419,12 +1734,17 @@ impl MitoHttpClient { suite_uuid: Uuid, overrides: HashMap, ) -> crate::error::Result { + let credential = self + .credential_guard + .get_credential() + .ok_or(CredentialGuardError::CredentialNotFound)? + .token; self.url .set_path(&format!("suites/{suite_uuid}/agents/override")); let resp = self .http_client .post(self.url.as_str()) - .bearer_auth(&self.credential) + .bearer_auth(credential) .json(&SuiteAgentOverrideReq { overrides }) .send() .await diff --git a/netmito/src/client/mod.rs b/netmito/src/client/mod.rs index 11461388..b8180058 100644 --- a/netmito/src/client/mod.rs +++ b/netmito/src/client/mod.rs @@ -10,7 +10,7 @@ use crate::{ config::{client::*, ClientConfig, ClientConfigCli}, entity::state::TaskExecState, schema::*, - service::auth::fill_user_login, + service::auth::{fill_user_login, get_and_prompt_username}, }; pub mod http; @@ -40,6 +40,8 @@ impl MitoClient { Ok(config) => match Self::setup(config).await { Ok(mut client) => { if let Some(cmd) = cli.command.take() { + // We ignore the return value here, the user can still re-login in interactive mode even if they called revoke + // in this subcommand. client.handle_command(cmd).await; } if cli.interactive { @@ -82,15 +84,12 @@ impl MitoClient { pub async fn setup(config: ClientConfig) -> crate::error::Result { tracing::debug!("Client is setting up"); - let mut http_client = MitoHttpClient::new(config.coordinator_addr); + let mut http_client = + MitoHttpClient::new(config.coordinator_addr, config.credential_path).await?; let username = http_client - .connect( - config.credential_path, - config.user, - config.password, - config.retain, - ) + .connect(config.user, config.password, config.refresh) .await?; + Ok(MitoClient { http_client, username, @@ -289,8 +288,23 @@ impl MitoClient { } pub async fn user_login(&mut self, args: LoginArgs) -> crate::error::Result<()> { - let req = fill_user_login(args.username, args.password, args.retain)?; - self.http_client.user_login(req).await + let username = match args.username { + Some(name) => name, + None => get_and_prompt_username(None, "Please input username")?, + }; + let req = fill_user_login(username, args.password, args.refresh)?; + let username = req.username.clone(); + self.http_client.user_login(req).await?; + self.username = username; + Ok(()) + } + + pub async fn revoke(&mut self) -> crate::error::Result<()> { + self.http_client.revoke().await + } + + pub async fn refresh_token(&mut self) -> crate::error::Result<()> { + self.http_client.refresh_token().await } pub async fn user_auth(&mut self) -> crate::error::Result { @@ -1409,6 +1423,23 @@ impl MitoClient { tracing::error!("{}", e); } }, + ClientCommand::Refresh => match self.refresh_token().await { + Ok(_) => { + tracing::info!("Successfully refreshed login token"); + } + Err(e) => { + tracing::error!("{}", e); + } + }, + ClientCommand::Revoke => match self.revoke().await { + Ok(_) => { + tracing::info!("Successfully revoked all login tokens"); + return false; + } + Err(e) => { + tracing::error!("{}", e); + } + }, ClientCommand::Login(args) => match self.user_login(args).await { Ok(_) => { tracing::info!("Successfully logged in as {}", self.username); diff --git a/netmito/src/config/client/mod.rs b/netmito/src/config/client/mod.rs index ed65d3ce..0d221916 100644 --- a/netmito/src/config/client/mod.rs +++ b/netmito/src/config/client/mod.rs @@ -45,7 +45,7 @@ pub struct ClientConfig { pub user: Option, pub password: Option, #[serde(default)] - pub retain: bool, + pub refresh: bool, } #[derive(Args, Debug, Serialize, Default, Clone)] @@ -74,10 +74,10 @@ pub struct ClientConfigCli { /// Enable interactive mode #[arg(short, long)] pub interactive: bool, - /// Whether to retain the previous login state without refetching the credential + /// Refresh current login token and invalidate previous tokens during client setup #[arg(long)] #[serde(skip_serializing_if = "<&bool>::not")] - pub retain: bool, + pub refresh: bool, /// The command to run #[command(subcommand)] #[serde(skip_serializing_if = "::std::option::Option::is_none")] @@ -97,6 +97,10 @@ pub enum ClientCommand { Admin(AdminArgs), /// Authenticate current user Auth, + /// Refresh current login token and invalidate previous tokens + Refresh, + /// Revoke all login tokens of current user + Revoke, /// Login with username and password Login(LoginArgs), /// Manage users, including changing password, querying the accessible groups etc. @@ -122,9 +126,10 @@ pub struct LoginArgs { pub username: Option, /// The password of the user pub password: Option, - /// Whether to retain the previous login state without refetching the credential + /// Refresh login state and invalidate previous tokens when logging in #[arg(long)] - pub retain: bool, + #[serde(default)] + pub refresh: bool, } #[derive(Serialize, Debug, Deserialize, Args, Clone)] @@ -250,7 +255,7 @@ impl Default for ClientConfig { credential_path: None, user: None, password: None, - retain: false, + refresh: false, } } } @@ -273,6 +278,7 @@ impl ClientConfig { .merge(Env::prefixed("MITO_").profile("client")) .merge(Serialized::from(cli, "client")) .select("client"); + Ok(figment.extract()?) } } diff --git a/netmito/src/config/worker.rs b/netmito/src/config/worker.rs index c169088a..40414ec7 100644 --- a/netmito/src/config/worker.rs +++ b/netmito/src/config/worker.rs @@ -129,11 +129,10 @@ pub struct WorkerConfig { pub(crate) file_log: bool, #[serde(default)] pub(crate) shared_log: bool, - #[serde(with = "humantime_serde")] + /// The lifetime of the worker token. `None` means the token never expires. + #[serde(default, with = "humantime_serde")] pub(crate) lifetime: Option, #[serde(default)] - pub(crate) retain: bool, - #[serde(default)] pub(crate) skip_redis: bool, } @@ -193,14 +192,13 @@ pub struct WorkerConfigCli { #[arg(long)] #[serde(skip_serializing_if = "<&bool>::not")] pub shared_log: bool, - /// The lifetime of the worker to alive (e.g., 7d, 1year) - #[arg(long)] - #[serde(skip_serializing_if = "::std::option::Option::is_none")] - pub lifetime: Option, - /// Whether to retain the previous login state without refetching the credential - #[arg(long)] - #[serde(skip_serializing_if = "<&bool>::not")] - pub retain: bool, + /// The lifetime of the worker token (e.g., 7d, 1year). If not given, the worker token is valid forever + #[arg(long, value_parser = humantime_serde::re::humantime::parse_duration)] + #[serde( + with = "humantime_serde", + skip_serializing_if = "::std::option::Option::is_none" + )] + pub lifetime: Option, /// Whether to skip connecting to Redis #[arg(long)] #[serde(skip_serializing_if = "<&bool>::not")] @@ -223,7 +221,6 @@ impl Default for WorkerConfig { file_log: false, shared_log: false, lifetime: None, - retain: false, skip_redis: false, } } diff --git a/netmito/src/error.rs b/netmito/src/error.rs index a166f6ce..2e3b0c10 100644 --- a/netmito/src/error.rs +++ b/netmito/src/error.rs @@ -57,6 +57,8 @@ pub enum Error { ParseSizeError(#[from] parse_size::Error), #[error("Parse int error: {0}")] ParseIntError(#[from] ParseIntError), + #[error("Credential guard error: {0}")] + CredentialGuardError(#[from] CredentialGuardError), } #[derive(thiserror::Error, Debug)] @@ -139,6 +141,12 @@ pub enum ApiError { PresignS3Error(Box), } +#[derive(thiserror::Error, Debug)] +pub enum CredentialGuardError { + #[error("Credential not found")] + CredentialNotFound, +} + #[derive(Serialize, Debug, Deserialize, Clone)] pub struct ErrorMsg { pub msg: String, diff --git a/netmito/src/manager.rs b/netmito/src/manager.rs index 6cf3535c..2b73bdcf 100644 --- a/netmito/src/manager.rs +++ b/netmito/src/manager.rs @@ -139,11 +139,9 @@ impl MitoManager { if worker_config.shared_log { cmd.arg("--shared-log"); } - if let Some(lifetime) = &worker_config.lifetime { - cmd.arg("--lifetime").arg(lifetime); - } - if worker_config.retain { - cmd.arg("--retain"); + if let Some(lifetime) = worker_config.lifetime { + cmd.arg("--lifetime") + .arg(humantime_serde::re::humantime::format_duration(lifetime).to_string()); } if worker_config.skip_redis { cmd.arg("--skip-redis"); diff --git a/netmito/src/schema/user.rs b/netmito/src/schema/user.rs index 2a55b0fa..dc26042b 100644 --- a/netmito/src/schema/user.rs +++ b/netmito/src/schema/user.rs @@ -9,12 +9,16 @@ pub struct CreateUserReq { pub admin: bool, } +fn default_refresh() -> bool { + false +} + #[derive(Debug, Serialize, Deserialize, Clone)] pub struct UserLoginReq { pub username: String, pub md5_password: [u8; 16], - #[serde(default)] - pub retain: bool, + #[serde(default = "default_refresh")] + pub refresh: bool, } #[derive(Debug, Serialize, Deserialize, Clone)] diff --git a/netmito/src/schema/worker.rs b/netmito/src/schema/worker.rs index 61ce6044..ab94ae23 100644 --- a/netmito/src/schema/worker.rs +++ b/netmito/src/schema/worker.rs @@ -12,8 +12,8 @@ pub struct RegisterWorkerReq { pub tags: HashSet, pub labels: HashSet, pub groups: HashSet, - #[serde(default)] - #[serde(with = "humantime_serde")] + /// The lifetime of the worker token. `None` means the token never expires. + #[serde(default, with = "humantime_serde")] pub lifetime: Option, } diff --git a/netmito/src/service/auth/cred.rs b/netmito/src/service/auth/cred.rs index e65b2363..47e1f267 100644 --- a/netmito/src/service/auth/cred.rs +++ b/netmito/src/service/auth/cred.rs @@ -3,25 +3,13 @@ use std::path::PathBuf; use figment::value::magic::RelativePathBuf; use reqwest::Client; -use tokio::io::AsyncBufReadExt; use url::Url; use crate::{ error::{ApiError, Error, ErrorMsg, RequestError}, - schema::UserLoginReq, - service::auth::fill_user_login, + service::auth::{credential_guard::CredentialGuard, fill_user_login}, }; -macro_rules! expect_two { - ($iter:expr) => {{ - let mut i = $iter; - match (i.next(), i.next(), i.next()) { - (Some(first), Some(second), None) => Some((first, second)), - _ => None, - } - }}; -} - pub trait GetPathBuf { fn get_path_buf(&self) -> PathBuf; } @@ -73,136 +61,28 @@ impl GetPathBuf for std::path::Path { // } // } -async fn read_lines

( - filename: P, -) -> std::io::Result>> -where - P: AsRef, -{ - let file = tokio::fs::File::open(filename).await?; - Ok(tokio::io::BufReader::new(file).lines()) -} - -async fn extract_credential( - user: Option<&String>, - lines: &mut tokio::io::Lines>, -) -> std::io::Result> { - match user { - // Specify the user, let us try to find the credential for the user - Some(user) => { - let prefix = format!("{user}:"); - while let Some(line) = lines.next_line().await? { - if line.starts_with(&prefix) { - if let Some((username, token)) = expect_two!(line.splitn(2, ':')) { - return Ok(Some((username.to_owned(), token.to_owned()))); - } - } - } - Ok(None) - } - // No user specified, just use the first line - None => { - if let Some(line) = lines.next_line().await? { - if let Some((username, token)) = expect_two!(line.splitn(2, ':')) { - return Ok(Some((username.to_owned(), token.to_owned()))); - } - } - Ok(None) - } - } -} - -// TODO: we might upgrade our credential storage format to include the coordinator address to avoid -// conflict -pub(crate) async fn modify_or_append_credential( - cred_path: &std::path::PathBuf, - username: &String, - token: &String, -) -> std::io::Result<()> { - if cred_path.exists() { - let mut lines = read_lines(cred_path).await?; - let mut new_lines = Vec::new(); - let prefix = format!("{username}:"); - let mut found = false; - while let Some(line) = lines.next_line().await? { - if line.starts_with(&prefix) { - new_lines.push(format!("{username}:{token}")); - found = true; - } else { - new_lines.push(line); - } - } - if !found { - new_lines.push(format!("{username}:{token}")); - } - tokio::fs::write(cred_path, new_lines.join("\n")).await?; - } else { - tokio::fs::write(cred_path, format!("{username}:{token}")).await?; - } - Ok(()) -} - // The return value is a tuple of username and token -pub async fn get_user_credential( - cred_path: Option<&RelativePathBuf>, +pub(crate) async fn get_user_credential( + credential_guard: &mut CredentialGuard, client: &Client, mut url: Url, - user: Option, + user: String, password: Option, - retain: bool, + refresh: bool, ) -> crate::error::Result<(String, String)> { - // Try to load credential from file - let cred_path = cred_path - .map(|p| p.relative()) - .or_else(|| { - dirs::config_dir().map(|mut p| { - p.push("mitosis"); - p.push("credentials"); - p - }) - }) - .ok_or(Error::ConfigError(Box::new(figment::Error::from( - "credential path not found", - ))))?; - // Check if the credential is valid - if cred_path.exists() { - if let Ok(mut lines) = read_lines(&cred_path).await { - if let Some((username, cred)) = extract_credential(user.as_ref(), &mut lines).await? { - url.set_path("auth"); - let resp = client - .get(url.as_str()) - .bearer_auth(&cred) - .send() - .await - .map_err(|e| { - if e.is_request() && e.is_connect() { - url.set_path(""); - RequestError::ConnectionError(url.to_string()) - } else { - e.into() - } - })?; - if resp.status().is_success() { - let resp_name = resp.text().await.map_err(RequestError::from)?; - if resp_name == username { - return Ok((username, cred)); - } - } else if resp.status().is_server_error() { - return Err(ApiError::InternalServerError.into()); - } - } - } - } - // Local credential not found or invalid, need to login - tracing::warn!("Local credential not found or invalid, need to login"); - let req = fill_user_login(user, password, retain)?; - url.set_path("login"); - let resp = client - .post(url.as_str()) - .json(&req) - .send() + if let Some((username, cred)) = credential_guard + .load_credential(user.clone()) .await - .map_err(|e| { + .map(|cred| (cred.username, cred.token)) + { + let request = if refresh { + url.set_path("refresh"); + client.post(url.as_str()) + } else { + url.set_path("auth"); + client.get(url.as_str()) + }; + let resp = request.bearer_auth(&cred).send().await.map_err(|e| { if e.is_request() && e.is_connect() { url.set_path(""); RequestError::ConnectionError(url.to_string()) @@ -210,37 +90,33 @@ pub async fn get_user_credential( e.into() } })?; - if resp.status().is_success() { - let resp = resp - .json::() - .await - .map_err(RequestError::from)?; - let token = resp.token; - if let Some(parent) = cred_path.parent() { - tokio::fs::create_dir_all(parent).await?; + let status = resp.status(); + if status.is_success() { + if refresh { + let resp = resp + .json::() + .await + .map_err(RequestError::from)?; + let token = resp.token; + credential_guard.save_credential(None, &token).await; + return Ok((username, token)); + } else { + let resp_name = resp.text().await.map_err(RequestError::from)?; + if resp_name == username { + return Ok((username, cred)); + } + } + } else if status.is_server_error() { + return Err(ApiError::InternalServerError.into()); } - modify_or_append_credential(&cred_path, &req.username, &token).await?; - Ok((req.username, token)) - } else { - let resp = resp.json::().await.map_err(RequestError::from)?; - Err(Error::Custom(resp.msg)) } -} -// This function is currently nowhere used, but it is kept for future potential use -pub async fn refresh_user_credential( - cred_path: Option<&T>, - client: &Client, - url: &mut Url, - user_login: &UserLoginReq, -) -> crate::error::Result -where - T: GetPathBuf, -{ + tracing::warn!("Local credential not found or invalid, need to login"); + let req = fill_user_login(user, password, refresh)?; url.set_path("login"); let resp = client .post(url.as_str()) - .json(&user_login) + .json(&req) .send() .await .map_err(|e| { @@ -257,18 +133,53 @@ where .await .map_err(RequestError::from)?; let token = resp.token; - if let Some(cred_path) = cred_path { - let cred_path = cred_path.get_path_buf(); - if cred_path.exists() { - if let Some(parent) = cred_path.parent() { - tokio::fs::create_dir_all(parent).await?; - } - modify_or_append_credential(&cred_path, &user_login.username, &token).await?; - } - } - Ok(token) + credential_guard.save_credential(None, &token).await; + Ok((req.username, token)) } else { let resp = resp.json::().await.map_err(RequestError::from)?; Err(Error::Custom(resp.msg)) } } + +// This function is currently nowhere used, but it is kept for future potential use +// pub async fn refresh_user_credential( +// cred_path: Option<&T>, +// client: &Client, +// url: &mut Url, +// user_login: &UserLoginReq, +// ) -> crate::error::Result +// where +// T: GetPathBuf, +// { +// url.set_path("login"); +// let resp = client +// .post(url.as_str()) +// .json(&user_login) +// .send() +// .await +// .map_err(|e| { +// if e.is_request() && e.is_connect() { +// url.set_path(""); +// RequestError::ConnectionError(url.to_string()) +// } else { +// e.into() +// } +// })?; +// if resp.status().is_success() { +// let resp = resp +// .json::() +// .await +// .map_err(RequestError::from)?; +// let token = resp.token; +// if let Some(cred_path) = cred_path { +// let credential_guard = CredentialStore::new(Some(cred_path.get_path_buf()))?; +// credential_guard +// .write_jwt(&*url, &user_login.username, &token) +// .await?; +// } +// Ok(token) +// } else { +// let resp = resp.json::().await.map_err(RequestError::from)?; +// Err(Error::Custom(resp.msg)) +// } +// } diff --git a/netmito/src/service/auth/credential_guard.rs b/netmito/src/service/auth/credential_guard.rs new file mode 100644 index 00000000..a6eaffad --- /dev/null +++ b/netmito/src/service/auth/credential_guard.rs @@ -0,0 +1,273 @@ +use std::path::PathBuf; + +use tokio::{fs, io}; +use url::Url; + +pub(crate) struct CredentialGuard { + /// `None` when no location could be resolved at all, in which case credentials + /// are only stored in-memory + credential_path: Option, + origin: String, + username: Option, + credential: Option, +} + +#[derive(Clone)] +pub(crate) struct ParsedCredential { + pub(crate) origin: String, + pub(crate) username: String, + pub(crate) token: String, +} + +fn normalize_origin(coordinator_url: &Url) -> String { + coordinator_url.origin().ascii_serialization() +} + +fn parse_credential(line: &str) -> Option { + let mut fields = line.split(','); + let (Some(origin), Some(username), Some(token), None) = + (fields.next(), fields.next(), fields.next(), fields.next()) + else { + return None; + }; + + Some(ParsedCredential { + origin: origin.to_string(), + username: username.to_string(), + token: token.to_string(), + }) +} + +impl CredentialGuard { + /// The caller creates the instance. We try to resolve the credential path. + /// + /// If the file isn't fully read/write-able, we warn it, but we will still record user's token + /// in-memory in self.active_credential. + /// + /// Upon new(), we will immediately try to parse the file and try to load the credential into + /// self.active_credential. If the credential is not found, the active_credential just stays + /// `None`. + pub(crate) async fn new(credential_path: Option, coordinator_url: &Url) -> Self { + let credential_path = credential_path.or_else(|| { + dirs::config_dir().map(|mut path| { + path.push("mitosis"); + path.push("credentials"); + path + }) + }); + + let mut credential_guard = Self { + credential_path, + origin: normalize_origin(coordinator_url), + username: None, + credential: None, + }; + + credential_guard.credential = credential_guard.load_credential_file().await; + + credential_guard + } + + /// The caller wants to access the credential for current username + pub(crate) fn get_credential(&self) -> Option { + match (&self.username, &self.credential) { + (Some(name), Some(cred)) => Some(ParsedCredential { + origin: self.origin.clone(), + username: name.clone(), + token: cred.clone(), + }), + _ => None, + } + } + + /// The caller wants to switch to this username. The returned value is + /// the stored credential for the new user. + pub(crate) async fn load_credential(&mut self, username: String) -> Option { + // Reloading for the active user would drop a credential that we failed to persist. + match &self.username { + Some(name) if *name == username => {} + _ => { + self.username = Some(username); + self.credential = self.load_credential_file().await; + } + } + + self.get_credential() + } + + /// The caller wants to update the active credential for current user + pub(crate) async fn save_credential(&mut self, username: Option, cred: &str) { + if let Err(error) = self.update_credential_file(username, Some(cred)).await { + tracing::warn!("Failed to save the credential: {error}."); + } + + self.credential = Some(cred.to_string()); + } + + /// The caller wants to drop the credential of the current user. It is forgotten in memory + /// whether or not the credential file could be updated. + pub(crate) async fn remove_credential(&mut self, username: Option) { + if let Err(error) = self.update_credential_file(username, None).await { + tracing::warn!("Failed to delete the credential: {error}."); + } + + self.credential = None; + } + + /// Set the credential of the active user in the credential file to `token`, or drop it when + /// there is no token. The file and its directory are created when a credential is written; a + /// credential that is not stored in the first place, like a store with no file at all, needs + /// no rewrite. + async fn update_credential_file( + &self, + username: Option, + token: Option<&str>, + ) -> io::Result<()> { + let username = match username.as_ref().or(self.username.as_ref()) { + Some(name) => name, + None => { + tracing::error!("Username not set for manipulating credential file"); + return Ok(()); + } + }; + let Some(credential_path) = self.credential_path.as_deref() else { + return Ok(()); + }; + + let mut lines: Vec = match fs::read_to_string(credential_path).await { + Ok(contents) => contents.lines().map(str::to_string).collect(), + Err(error) if error.kind() == io::ErrorKind::NotFound => Vec::new(), + Err(error) => return Err(error), + }; + + let stored = lines.iter().position(|line| { + parse_credential(line) + .is_some_and(|stored| stored.origin == self.origin && stored.username == *username) + }); + + match (stored, token) { + (Some(index), Some(token)) => { + lines[index] = format!("{},{},{}", self.origin, username, token) + } + (Some(index), None) => { + lines.remove(index); + } + (None, Some(token)) => lines.push(format!("{},{},{}", self.origin, username, token)), + // There is no such credential to remove, so the file needs no rewrite. + (None, None) => return Ok(()), + } + + if let Some(parent) = credential_path + .parent() + .filter(|parent| !parent.as_os_str().is_empty()) + { + fs::create_dir_all(parent).await?; + } + + let mut contents = lines.join("\n"); + if !contents.is_empty() { + contents.push('\n'); + } + + fs::write(credential_path, contents).await + } + + /// Look up the stored credential of the active user. A file we cannot read is reported and + /// then treated as if it held no matching credential. + async fn load_credential_file(&self) -> Option { + let username = match &self.username { + Some(name) => name, + None => { + tracing::error!("Username not set for loading credential"); + return None; + } + }; + let credential_path = self.credential_path.as_deref()?; + + let contents = match fs::read_to_string(credential_path).await { + Ok(contents) => contents, + Err(error) => { + if error.kind() != io::ErrorKind::NotFound { + tracing::warn!( + "Failed to read credential file {}: {error}", + credential_path.display() + ); + } + return None; + } + }; + + contents + .lines() + .filter_map(parse_credential) + .find(|credential| credential.origin == self.origin && credential.username == *username) + .map(|cred| cred.token) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn normalized_origin(url: &str) -> String { + normalize_origin(&Url::parse(url).expect("test URL should be valid")) + } + + #[test] + fn normalize_equivalent_http_origins() { + for url in [ + "http://127.0.0.1/", + "http://127.0.0.1", + "http://127.0.0.1:80", + "http://127.0.0.1:80/", + ] { + assert_eq!(normalized_origin(url), "http://127.0.0.1"); + } + } + + #[test] + fn normalize_origin_preserves_identity_differences() { + assert_eq!( + normalized_origin("https://example.com:443/"), + "https://example.com" + ); + assert_eq!( + normalized_origin("http://example.com:5000/"), + "http://example.com:5000" + ); + assert_ne!( + normalized_origin("http://example.com"), + normalized_origin("https://example.com") + ); + assert_ne!( + normalized_origin("http://example.com"), + normalized_origin("http://example.org") + ); + assert_eq!( + normalized_origin("http://example.com/path?q=1#section"), + "http://example.com" + ); + assert_ne!( + normalized_origin("http://example.com:5000"), + normalized_origin("http://example.com:5001") + ); + assert_ne!( + normalized_origin("http://localhost"), + normalized_origin("http://127.0.0.1") + ); + } + + #[test] + fn parse_comma_separated_credential() { + let credential = parse_credential("http://127.0.0.1,user_a,encoded-token==") + .expect("credential should be valid"); + + assert_eq!(credential.origin, "http://127.0.0.1"); + assert_eq!(credential.username, "user_a"); + assert_eq!(credential.token, "encoded-token=="); + + assert!(parse_credential("user_a:legacy-token").is_none()); + assert!(parse_credential("http://127.0.0.1,user_a").is_none()); + assert!(parse_credential("http://127.0.0.1,user_a,token,extra").is_none()); + } +} diff --git a/netmito/src/service/auth/mod.rs b/netmito/src/service/auth/mod.rs index cfc1171b..20985ad9 100644 --- a/netmito/src/service/auth/mod.rs +++ b/netmito/src/service/auth/mod.rs @@ -1,4 +1,5 @@ pub mod cred; +pub(crate) mod credential_guard; pub mod token; use std::{io::Write, net::SocketAddr}; @@ -49,6 +50,19 @@ pub struct AuthWorker { pub uuid: Uuid, } +fn generate_auth_signature() -> i64 { + StdRng::from_os_rng().next_u32() as i64 + 1 +} + +fn generate_new_auth_signature(old: Option) -> i64 { + loop { + let sign = generate_auth_signature(); + if Some(sign) != old { + return sign; + } + } +} + pub(crate) fn get_and_prompt_username( username: Option, prompt: &str, @@ -75,7 +89,7 @@ pub(crate) fn get_and_prompt_password( ) -> crate::error::Result<[u8; 16]> { let md5_password = password .map(|p| { - println!("{prompt} Already Given"); + // println!("{prompt} Already Given"); Ok::<_, std::io::Error>(md5::compute(p.as_bytes()).0) }) .unwrap_or_else(|| { @@ -86,33 +100,26 @@ pub(crate) fn get_and_prompt_password( } pub(crate) fn fill_user_login( - username: Option, + username: String, password: Option, - retain: bool, + refresh: bool, ) -> crate::error::Result { - match (username, password) { - (Some(username), Some(password)) => Ok(UserLoginReq { - username, - md5_password: md5::compute(password.as_bytes()).0, - retain, - }), - (username, password) => { - let username = get_and_prompt_username(username, "Username")?; - let md5_password = get_and_prompt_password(password, "Password")?; - Ok(UserLoginReq { - username, - md5_password, - retain, - }) - } - } + let md5_password = match password { + Some(password) => md5::compute(password.as_bytes()).0, + None => get_and_prompt_password(password, &format!("Password for {}", username))?, + }; + Ok(UserLoginReq { + username, + md5_password, + refresh, + }) } pub async fn user_login( db: &DatabaseConnection, username: &str, md5_password: &[u8; 16], - retain: bool, + refresh: bool, ip: SocketAddr, ) -> crate::error::Result { match User::Entity::find() @@ -129,11 +136,10 @@ pub async fn user_login( .verify_password(md5_password, &parsed_hash) .is_ok() { - let sign = if retain { - user.auth_signature - .unwrap_or_else(|| StdRng::from_os_rng().next_u32() as i64) + let sign = if refresh { + generate_new_auth_signature(user.auth_signature) } else { - (1 + StdRng::from_os_rng().next_u32()) as i64 + user.auth_signature.unwrap_or_else(generate_auth_signature) }; let token = generate_token(username, sign)?; let now = TimeDateTimeWithTimeZone::now_utc(); @@ -162,6 +168,61 @@ pub async fn user_login( } } +pub async fn refresh_user_token( + db: &DatabaseConnection, + user_id: i64, + ip: SocketAddr, +) -> crate::error::Result { + let user = User::Entity::find_by_id(user_id) + .one(db) + .await? + .ok_or(ApiError::NotFound("User not found".to_string()))?; + + if user.state != UserState::Active { + return Err(AuthError::PermissionDenied.into()); + } + + let sign = generate_new_auth_signature(user.auth_signature); + let token = generate_token(&user.username, sign)?; + let now = TimeDateTimeWithTimeZone::now_utc(); + + let active_user = User::ActiveModel { + id: Set(user.id), + auth_signature: Set(Some(sign)), + current_sign_in_at: Set(Some(now)), + last_sign_in_at: Set(user.current_sign_in_at), + current_sign_in_ip: Set(Some(ip.ip().to_string())), + last_sign_in_ip: Set(user.current_sign_in_ip), + updated_at: Set(now), + ..Default::default() + }; + + active_user.update(db).await?; + Ok(token) +} + +pub async fn revoke(db: &DatabaseConnection, user_id: i64) -> crate::error::Result<()> { + let user = User::Entity::find_by_id(user_id) + .one(db) + .await? + .ok_or(ApiError::NotFound("User not found".to_string()))?; + + if user.state != UserState::Active { + return Err(AuthError::PermissionDenied.into()); + } + + let now = TimeDateTimeWithTimeZone::now_utc(); + let active_user = User::ActiveModel { + id: Set(user.id), + auth_signature: Set(Some(generate_new_auth_signature(user.auth_signature))), + updated_at: Set(now), + ..Default::default() + }; + + active_user.update(db).await?; + Ok(()) +} + pub async fn user_change_password( db: &DatabaseConnection, user_id: i64, @@ -268,9 +329,8 @@ pub async fn user_auth_with_name_middleware( async fn user_auth(db: &DatabaseConnection, bearer: &Bearer) -> Result { let token = bearer.token(); let claims = verify_token(token).map_err(|_| AuthError::InvalidToken)?; - let now = TimeDateTimeWithTimeZone::now_utc(); - if claims.exp < now { - return Err(AuthError::WrongCredentials); + if claims.exp.is_none() { + return Err(AuthError::InvalidToken); } let user = User::Entity::find() @@ -303,9 +363,8 @@ pub async fn admin_auth_middleware( async fn admin_auth(db: &DatabaseConnection, bearer: &Bearer) -> Result { let token = bearer.token(); let claims = verify_token(token).map_err(|_| AuthError::InvalidToken)?; - let now = TimeDateTimeWithTimeZone::now_utc(); - if claims.exp < now { - return Err(AuthError::WrongCredentials); + if claims.exp.is_none() { + return Err(AuthError::InvalidToken); } let user = User::Entity::find() diff --git a/netmito/src/service/auth/token.rs b/netmito/src/service/auth/token.rs index 27f63461..730fe290 100644 --- a/netmito/src/service/auth/token.rs +++ b/netmito/src/service/auth/token.rs @@ -12,8 +12,12 @@ pub struct TokenClaims<'a> { /// username pub sub: Cow<'a, str>, /// expiry time - #[serde(with = "jwt_numeric_date")] - pub exp: OffsetDateTime, + #[serde( + default, + skip_serializing_if = "Option::is_none", + with = "jwt_numeric_date_opt" + )] + pub exp: Option, /// random number pub sign: i64, } @@ -29,7 +33,7 @@ where ))?; let claims = TokenClaims { sub: Cow::from(username.as_ref()), - exp: OffsetDateTime::now_utc() + token_ttl.token_expires_in, + exp: Some(OffsetDateTime::now_utc() + token_ttl.token_expires_in), sign, }; @@ -41,6 +45,7 @@ where encode_token(&claims, encoding_key) } +/// Generate a worker token expiring after `lifetime`, or one that never expires if it is `None`. pub fn generate_worker_token( username: T, sign: i64, @@ -49,25 +54,20 @@ pub fn generate_worker_token( where T: AsRef, { - let token_ttl = match lifetime { - Some(ttl) => time::Duration::try_from(ttl).map_err(|_| { - ApiError::InvalidRequest(format!( - "Invalid lifetime {}", - humantime_serde::re::humantime::format_duration(ttl) - )) - })?, - None => { - crate::config::SERVER_CONFIG - .get() - .ok_or(crate::error::Error::Custom( - "server config not found".to_string(), - ))? - .token_expires_in - } - }; + let exp = lifetime + .map(|ttl| { + let token_ttl = time::Duration::try_from(ttl).map_err(|_| { + ApiError::InvalidRequest(format!( + "Invalid lifetime {}", + humantime_serde::re::humantime::format_duration(ttl) + )) + })?; + Ok::<_, ApiError>(OffsetDateTime::now_utc() + token_ttl) + }) + .transpose()?; let claims = TokenClaims { sub: Cow::from(username.as_ref()), - exp: OffsetDateTime::now_utc() + token_ttl, + exp, sign, }; @@ -94,7 +94,8 @@ pub fn verify_token(token: &str) -> crate::error::Result> { .decode(token) .map_err(DecodeTokenError::from)?; let token = String::from_utf8(token).map_err(DecodeTokenError::from)?; - let validation = jsonwebtoken::Validation::new(jsonwebtoken::Algorithm::EdDSA); + let mut validation = jsonwebtoken::Validation::new(jsonwebtoken::Algorithm::EdDSA); + validation.required_spec_claims.remove("exp"); let decoding_key = crate::config::DECODING_KEY .get() .ok_or(crate::error::Error::Custom( @@ -105,24 +106,119 @@ pub fn verify_token(token: &str) -> crate::error::Result> { Ok(decoded.claims) } -mod jwt_numeric_date { +#[cfg(test)] +mod tests { + use super::*; + use jsonwebtoken::{Algorithm, DecodingKey, EncodingKey, Validation}; + + // Ed25519 test keypair, local to this module only. Never touches + // `crate::config::ENCODING_KEY`/`DECODING_KEY` (process-global `OnceCell`s) so these + // tests carry no shared state and can't affect or be affected by other tests. + const TEST_PRIVATE_KEY_PEM: &[u8] = b"-----BEGIN PRIVATE KEY-----\n\ +MC4CAQAwBQYDK2VwBCIEIE4+s/pbE45hkcW3aMQOwcwQdGsE8fWZ6Zr8PbKHYtHa\n\ +-----END PRIVATE KEY-----\n"; + const TEST_PUBLIC_KEY_PEM: &[u8] = b"-----BEGIN PUBLIC KEY-----\n\ +MCowBQYDK2VwAyEAa+75592cq25dhxP5a9zZMvLn+7yXyq6cUj2xmjB2PW4=\n\ +-----END PUBLIC KEY-----\n"; + + fn test_encoding_key() -> EncodingKey { + EncodingKey::from_ed_pem(TEST_PRIVATE_KEY_PEM).expect("valid test private key") + } + + fn test_decoding_key() -> DecodingKey { + DecodingKey::from_ed_pem(TEST_PUBLIC_KEY_PEM).expect("valid test public key") + } + + fn encode_test_claims(exp: Option) -> String { + let claims = TokenClaims { + sub: Cow::from("test_user"), + exp, + sign: 42, + }; + encode_token(&claims, &test_encoding_key()).expect("token should encode") + } + + // Learning test helper: mirrors `verify_token`'s decode/validation logic exactly + // (same base64 unwrap + same `Validation` setup: EdDSA, "exp" removed from + // `required_spec_claims`), but takes the decoding key as a parameter instead of + // reading the global `crate::config::DECODING_KEY`. Keep in sync with `verify_token` + // above if its validation setup ever changes. + fn decode_test_claims<'a>( + token: &'a str, + decoding_key: &DecodingKey, + ) -> crate::error::Result> { + let token = general_purpose::STANDARD + .decode(token) + .map_err(DecodeTokenError::from)?; + let token = String::from_utf8(token).map_err(DecodeTokenError::from)?; + let mut validation = Validation::new(Algorithm::EdDSA); + validation.required_spec_claims.remove("exp"); + let decoded = jsonwebtoken::decode::(&token, decoding_key, &validation) + .map_err(DecodeTokenError::from)?; + Ok(decoded.claims) + } + + #[test] + fn verify_token_accepts_missing_exp() { + let token = encode_test_claims(None); + + let claims = decode_test_claims(&token, &test_decoding_key()) + .expect("token without exp should verify"); + + assert_eq!(claims.sub, "test_user"); + assert_eq!(claims.sign, 42); + assert!(claims.exp.is_none()); + } + + #[test] + fn verify_token_accepts_valid_future_exp() { + let exp = OffsetDateTime::now_utc() + time::Duration::hours(1); + let token = encode_test_claims(Some(exp)); + + let claims = decode_test_claims(&token, &test_decoding_key()) + .expect("token with future exp should verify"); + + assert!(claims.exp.is_some()); + assert_eq!(claims.exp.unwrap().unix_timestamp(), exp.unix_timestamp()); + } + + #[test] + fn verify_token_rejects_expired_exp() { + let exp = OffsetDateTime::now_utc() - time::Duration::hours(1); + let token = encode_test_claims(Some(exp)); + + let result = decode_test_claims(&token, &test_decoding_key()); + + assert!( + result.is_err(), + "expired token should be rejected, got {result:?}" + ); + } +} + +mod jwt_numeric_date_opt { use serde::{self, Deserialize, Deserializer, Serializer}; use time::OffsetDateTime; - /// Serializes an OffsetDateTime to a Unix timestamp (milliseconds since 1970/1/1T00:00:00T) - pub fn serialize(date: &OffsetDateTime, serializer: S) -> Result + + pub fn serialize(date: &Option, serializer: S) -> Result where S: Serializer, { - let timestamp = date.unix_timestamp(); - serializer.serialize_i64(timestamp) + match date { + Some(date) => serializer.serialize_some(&date.unix_timestamp()), + None => serializer.serialize_none(), + } } - /// Attempts to deserialize an i64 and use as a Unix timestamp - pub fn deserialize<'de, D>(deserializer: D) -> Result + pub fn deserialize<'de, D>(deserializer: D) -> Result, D::Error> where D: Deserializer<'de>, { - OffsetDateTime::from_unix_timestamp(i64::deserialize(deserializer)?) - .map_err(|_| serde::de::Error::custom("invalid Unix timestamp value")) + Option::::deserialize(deserializer)? + .map(|timestamp| { + OffsetDateTime::from_unix_timestamp(timestamp) + .map_err(|_| serde::de::Error::custom("invalid Unix timestamp value")) + }) + .transpose() } } diff --git a/netmito/src/worker.rs b/netmito/src/worker.rs index b9607eb3..39527a39 100644 --- a/netmito/src/worker.rs +++ b/netmito/src/worker.rs @@ -27,12 +27,13 @@ use crate::entity::content::ArtifactContentType; use crate::entity::state::TaskExecState; use crate::error::RequestError; use crate::schema::*; +use crate::service::auth::get_and_prompt_username; use crate::service::s3::download_file; use crate::{ config::{WorkerConfig, WorkerConfigCli}, error::{Error, ErrorMsg}, schema::{RegisterWorkerReq, RegisterWorkerResp}, - service::auth::cred::get_user_credential, + service::auth::{cred::get_user_credential, credential_guard::CredentialGuard}, signal::shutdown_signal, }; @@ -251,11 +252,23 @@ impl MitoWorker { pub async fn setup(mut config: WorkerConfig) -> crate::error::Result<(Self, TracingGuard)> { tracing::debug!("Worker is setting up"); let http_client = Client::new(); + let mut credential_guard = CredentialGuard::new( + config + .credential_path + .as_ref() + .map(|credential_path| credential_path.relative()), + &config.coordinator_addr, + ) + .await; + let username = match &config.user { + Some(name) => name.to_string(), + None => get_and_prompt_username(None, "Please input username")?, + }; let (_, credential) = get_user_credential( - config.credential_path.as_ref(), + &mut credential_guard, &http_client, config.coordinator_addr.clone(), - config.user.take(), + username, config.password.take(), false, ) diff --git a/openapi.yaml b/openapi.yaml index b448a943..8f218fb2 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -12,6 +12,8 @@ info: ## API Structure - `/auth` - User authentication status - `/login` - User login + - `/refresh` - Refresh current login token + - `/revoke` - Revoke all login tokens for current user - `/users` - User management - `/groups` - Group management and file attachments - `/workers` - Worker registration and management @@ -78,6 +80,40 @@ paths: "400": $ref: "#/components/responses/BadRequest" + /refresh: + post: + summary: Refresh current login token + description: Refresh the current user's login token using a valid bearer token. This invalidates previously issued tokens for the same user. + operationId: refreshToken + tags: + - Authentication + security: + - bearerAuth: [] + responses: + "200": + description: Token refreshed successfully + content: + application/json: + schema: + $ref: "#/components/schemas/UserLoginResp" + "401": + $ref: "#/components/responses/Unauthorized" + + /revoke: + post: + summary: Revoke all login tokens + description: Revoke all issued login tokens for the current authenticated user. + operationId: revokeTokens + tags: + - Authentication + security: + - bearerAuth: [] + responses: + "200": + description: Successfully revoked all login tokens + "401": + $ref: "#/components/responses/Unauthorized" + /auth: get: summary: Get current user @@ -2287,8 +2323,8 @@ components: description: MD5 hash of password as byte array retain: type: boolean - default: false - description: Whether to retain existing login state + default: true + description: Whether to retain existing login state. Defaults to true. Set to false to refresh the login state and invalidate previously issued tokens. UserLoginResp: type: object @@ -2836,7 +2872,7 @@ components: lifetime: type: string nullable: true - description: Worker lifetime (e.g. "7d", "1h") + description: Worker JWT token lifetime. Use a duration such as "7d", "1h", or "30m". Omit the field or set it to null to issue a token that never expires. The coordinator's access_token_expires_in does not apply to worker tokens. example: 7d RegisterWorkerResp: