Substitute credentials authentication by token authentication with auto-renewal

c/Cargo.toml

@@ -21,7 +21,7 @@ features = {}
 
 	[dependencies.log]
 		features = ["kv", "kv_unstable", "std", "value-bag"]
-		version = "0.4.26"
+		version = "0.4.27"
 		default-features = false
 
 	[dependencies.typedb-driver]

dependencies/typedb/repositories.bzl

@@ -18,22 +18,25 @@
 load("@bazel_tools//tools/build_defs/repo:git.bzl", "git_repository")
 
 def typedb_dependencies():
+    # TODO: Return typedb
     git_repository(
         name = "typedb_dependencies",
-        remote = "https://github.com/typedb/typedb-dependencies",
-        commit = "cf9c1707c7896d61ff97bbf60b1880852ad42353",  # sync-marker: do not remove this comment, this is used for sync-dependencies by @typedb_dependencies
+        remote = "https://github.com/farost/typedb-dependencies",
+        commit = "033496f9ead5e0f7516a232dcc2b5e031415a36a",  # sync-marker: do not remove this comment, this is used for sync-dependencies by @typedb_dependencies
     )
 
 def typedb_protocol():
+    # TODO: Return typedb
     git_repository(
         name = "typedb_protocol",
-        remote = "https://github.com/typedb/typedb-protocol",
-        tag = "3.0.0",  # sync-marker: do not remove this comment, this is used for sync-dependencies by @typedb_protocol
+        remote = "https://github.com/farost/typedb-protocol",
+        commit = "7df6a6bfbbbd29940558ae7940b2065e1e2ba0b1",  # sync-marker: do not remove this comment, this is used for sync-dependencies by @typedb_protocol
     )
 
 def typedb_behaviour():
+    # TODO: Return typedb
     git_repository(
         name = "typedb_behaviour",
-        remote = "https://github.com/typedb/typedb-behaviour",
-        commit = "a5ca738d691e7e7abec0a69e68f6b06310ac2168",  # sync-marker: do not remove this comment, this is used for sync-dependencies by @typedb_behaviour
+        remote = "https://github.com/farost/typedb-behaviour",
+        commit = "54d190804025d71e2f2498a63ab508bf68c5f3c9",  # sync-marker: do not remove this comment, this is used for sync-dependencies by @typedb_behaviour
     )

rust/Cargo.toml

@@ -28,7 +28,7 @@
 
 	[dev-dependencies.async-std]
 		features = ["alloc", "async-attributes", "async-channel", "async-global-executor", "async-io", "async-lock", "attributes", "crossbeam-utils", "default", "futures-channel", "futures-core", "futures-io", "futures-lite", "gloo-timers", "kv-log-macro", "log", "memchr", "once_cell", "pin-project-lite", "pin-utils", "slab", "std", "wasm-bindgen-futures"]
-		version = "1.13.0"
+		version = "1.13.1"
 		default-features = false
 
 	[dev-dependencies.steps]
@@ -55,18 +55,18 @@
 
 	[dependencies.tokio]
 		features = ["bytes", "default", "fs", "full", "io-std", "io-util", "libc", "macros", "mio", "net", "parking_lot", "process", "rt", "rt-multi-thread", "signal", "signal-hook-registry", "socket2", "sync", "time", "tokio-macros"]
-		version = "1.43.0"
+		version = "1.44.1"
 		default-features = false
 
 	[dependencies.typedb-protocol]
 		features = []
+		rev = "7df6a6bfbbbd29940558ae7940b2065e1e2ba0b1"
 		git = "https://github.com/typedb/typedb-protocol"
-		tag = "3.0.0"
 		default-features = false
 
 	[dependencies.log]
 		features = ["kv", "kv_unstable", "std", "value-bag"]
-		version = "0.4.26"
+		version = "0.4.27"
 		default-features = false
 
 	[dependencies.tokio-stream]
@@ -81,7 +81,7 @@
 
 	[dependencies.uuid]
 		features = ["default", "fast-rng", "rng", "serde", "std", "v4"]
-		version = "1.15.1"
+		version = "1.16.0"
 		default-features = false
 
 	[dependencies.itertools]
@@ -106,7 +106,7 @@
 
 	[dependencies.http]
 		features = ["default", "std"]
-		version = "1.2.0"
+		version = "1.3.1"
 		default-features = false
 
 	[dependencies.maybe-async]

rust/src/common/error.rs

@@ -21,7 +21,7 @@ use std::{collections::HashSet, error::Error as StdError, fmt};
 
 use itertools::Itertools;
 use tonic::{Code, Status};
-use tonic_types::StatusExt;
+use tonic_types::{ErrorDetails, ErrorInfo, StatusExt};
 
 use super::{address::Address, RequestID};
 
@@ -150,7 +150,7 @@ error_messages! { ConnectionError
         15: "The replica is not the primary replica.",
     ClusterAllNodesFailed { errors: String } =
         16: "Attempted connecting to all TypeDB Cluster servers, but the following errors occurred: \n{errors}.",
-    ClusterTokenCredentialInvalid =
+    TokenCredentialInvalid =
         17: "Invalid token credentials.",
     EncryptionSettingsMismatch =
         18: "Unable to connect to TypeDB: possible encryption settings mismatch.",
@@ -275,6 +275,16 @@ impl Error {
         }
     }
 
+    fn try_extracting_connection_error(status: &Status, code: &str) -> Option<ConnectionError> {
+        // TODO: We should probably catch more connection errors instead of wrapping them into
+        // ServerErrors. However, the most valuable information even for connection is inside
+        // stacktraces now.
+        match code {
+            "AUT3" => Some(ConnectionError::TokenCredentialInvalid {}),
+            _ => None,
+        }
+    }
+
     fn from_message(message: &str) -> Self {
         // TODO: Consider converting some of the messages to connection errors
         Self::Other(message.to_owned())
@@ -352,9 +362,13 @@ impl From<Status> for Error {
                 })
             } else if let Some(error_info) = details.error_info() {
                 let code = error_info.reason.clone();
+                if let Some(connection_error) = Self::try_extracting_connection_error(&status, &code) {
+                    return Self::Connection(connection_error);
+                }
                 let domain = error_info.domain.clone();
                 let stack_trace =
                     if let Some(debug_info) = details.debug_info() { debug_info.stack_entries.clone() } else { vec![] };
+
                 Self::Server(ServerError::new(code, domain, status.message().to_owned(), stack_trace))
             } else {
                 Self::from_message(status.message())
@@ -364,7 +378,6 @@ impl From<Status> for Error {
                 Self::parse_unavailable(status.message())
             } else if status.code() == Code::Unknown
                 || is_rst_stream(&status)
-                || status.code() == Code::InvalidArgument
                 || status.code() == Code::FailedPrecondition
                 || status.code() == Code::AlreadyExists
             {

rust/src/connection/message.rs

@@ -35,12 +35,12 @@ use crate::{
     error::ServerError,
     info::UserInfo,
     user::User,
-    Options, TransactionType,
+    Credentials, Options, TransactionType,
 };
 
 #[derive(Debug)]
 pub(super) enum Request {
-    ConnectionOpen { driver_lang: String, driver_version: String },
+    ConnectionOpen { driver_lang: String, driver_version: String, credentials: Credentials },
 
     ServersAll,
 

rust/src/connection/network/channel.rs

@@ -17,11 +17,12 @@
  * under the License.
  */
 
-use std::sync::Arc;
+use std::sync::{Arc, RwLock};
 
 use tonic::{
     body::BoxBody,
     client::GrpcService,
+    metadata::MetadataValue,
     service::{
         interceptor::{InterceptedService, ResponseFuture as InterceptorResponseFuture},
         Interceptor,
@@ -65,20 +66,33 @@ pub(super) fn open_callcred_channel(
 #[derive(Debug)]
 pub(super) struct CallCredentials {
     credentials: Credentials,
+    token: RwLock<Option<String>>,
 }
 
 impl CallCredentials {
     pub(super) fn new(credentials: Credentials) -> Self {
-        Self { credentials }
+        Self { credentials, token: RwLock::new(None) }
     }
 
-    pub(super) fn username(&self) -> &str {
-        self.credentials.username()
+    pub(super) fn credentials(&self) -> &Credentials {
+        &self.credentials
+    }
+
+    pub(super) fn set_token(&self, token: String) {
+        *self.token.write().expect("Expected token write lock acquisition on set") = Some(token);
+    }
+
+    pub(super) fn reset_token(&self) {
+        *self.token.write().expect("Expected token write lock acquisition on reset") = None;
     }
 
     pub(super) fn inject(&self, mut request: Request<()>) -> Request<()> {
-        request.metadata_mut().insert("username", self.credentials.username().try_into().unwrap());
-        request.metadata_mut().insert("password", self.credentials.password().try_into().unwrap());
+        if let Some(token) = &*self.token.read().expect("Expected token read lock acquisition on inject") {
+            request.metadata_mut().insert(
+                "authorization",
+                format!("Bearer {}", token).try_into().expect("Expected authorization header formatting"),
+            );
+        }
         request
     }
 }

rust/src/connection/network/proto/message.rs

@@ -19,8 +19,8 @@
 
 use itertools::Itertools;
 use typedb_protocol::{
-    connection, database, database_manager, query::initial_res::Res, server_manager, transaction, user, user_manager,
-    Version::Version,
+    authentication, connection, database, database_manager, query::initial_res::Res, server_manager, transaction, user,
+    user_manager, Version::Version,
 };
 use uuid::Uuid;
 
@@ -32,14 +32,18 @@ use crate::{
     error::{ConnectionError, InternalError, ServerError},
     info::UserInfo,
     user::User,
+    Credentials,
 };
 
 impl TryIntoProto<connection::open::Req> for Request {
     fn try_into_proto(self) -> Result<connection::open::Req> {
         match self {
-            Self::ConnectionOpen { driver_lang, driver_version } => {
-                Ok(connection::open::Req { version: Version.into(), driver_lang, driver_version })
-            }
+            Self::ConnectionOpen { driver_lang, driver_version, credentials } => Ok(connection::open::Req {
+                version: Version.into(),
+                driver_lang,
+                driver_version,
+                authentication: Some(credentials.try_into_proto()?),
+            }),
             other => Err(InternalError::UnexpectedRequestType { request_type: format!("{other:?}") }.into()),
         }
     }
@@ -225,14 +229,28 @@ impl TryIntoProto<user::delete::Req> for Request {
     }
 }
 
+impl TryIntoProto<authentication::token::create::Req> for Credentials {
+    fn try_into_proto(self) -> Result<authentication::token::create::Req> {
+        Ok(authentication::token::create::Req {
+            credentials: Some(authentication::token::create::req::Credentials::Password(
+                authentication::token::create::req::Password {
+                    username: self.username().to_owned(),
+                    password: self.password().to_owned(),
+                },
+            )),
+        })
+    }
+}
+
 impl TryFromProto<connection::open::Res> for Response {
     fn try_from_proto(proto: connection::open::Res) -> Result<Self> {
         let mut database_infos = Vec::new();
-        for database_info_proto in proto.databases_all.unwrap().databases {
+        for database_info_proto in proto.databases_all.expect("Expected databases data").databases {
             database_infos.push(DatabaseInfo::try_from_proto(database_info_proto)?);
         }
         Ok(Self::ConnectionOpen {
-            connection_id: Uuid::from_slice(proto.connection_id.unwrap().id.as_slice()).unwrap(),
+            connection_id: Uuid::from_slice(proto.connection_id.expect("Expected connection id").id.as_slice())
+                .unwrap(),
             server_duration_millis: proto.server_duration_millis,
             databases: database_infos,
         })