aster_forge_auth/
password.rs

1use std::sync::Arc;
2
3use aster_forge_crypto::{
4    PasswordHashPolicy, PasswordHashVerification, hash_password_with_policy,
5    verify_password_with_policy,
6};
7use tokio::sync::Semaphore;
8
9use aster_forge_crypto::CryptoError;
10
11/// Redacted async password-work errors. Products map these at their own boundary.
12#[derive(Debug, thiserror::Error)]
13pub enum PasswordWorkError {
14    /// A zero-sized executor budget was requested.
15    #[error("password work concurrency must be positive")]
16    InvalidConcurrency,
17    /// No immediate execution permit is available.
18    #[error("password work budget exhausted")]
19    Busy,
20    /// The execution semaphore was closed.
21    #[error("password worker is closed")]
22    Closed,
23    /// The blocking job failed to join.
24    #[error("password worker failed")]
25    Worker,
26    /// The synchronous cryptographic primitive rejected the operation.
27    #[error("password cryptographic operation failed")]
28    Crypto(#[from] CryptoError),
29}
30type Result<T> = std::result::Result<T, PasswordWorkError>;
31
32/// Default number of concurrent memory-hard password jobs.
33pub const DEFAULT_PASSWORD_HASH_MAX_CONCURRENCY: usize = 2;
34
35/// Instance-owned blocking executor with a memory/CPU concurrency budget.
36#[derive(Clone, Debug)]
37pub struct PasswordHashRuntime {
38    policy: PasswordHashPolicy,
39    semaphore: Arc<Semaphore>,
40}
41
42impl PasswordHashRuntime {
43    /// Creates an executor using the Forge default password policy.
44    ///
45    /// # Errors
46    /// Returns `InvalidConcurrency` if the requested concurrency is zero.
47    pub fn new(max_concurrency: usize) -> Result<Self> {
48        Self::with_policy(max_concurrency, PasswordHashPolicy::default())
49    }
50
51    /// Creates an executor with an explicit policy and positive concurrency limit.
52    ///
53    /// # Errors
54    /// Returns `InvalidConcurrency` if the requested concurrency is zero.
55    pub fn with_policy(max_concurrency: usize, policy: PasswordHashPolicy) -> Result<Self> {
56        if max_concurrency == 0 {
57            return Err(PasswordWorkError::InvalidConcurrency);
58        }
59
60        Ok(Self {
61            policy,
62            semaphore: Arc::new(Semaphore::new(max_concurrency)),
63        })
64    }
65
66    /// Hashes a password on the blocking pool after acquiring an execution permit.
67    ///
68    /// # Errors
69    /// Returns an error when the worker is closed, fails, or cannot hash the input.
70    pub async fn hash_password(&self, password: &str) -> Result<String> {
71        let permit = self
72            .semaphore
73            .clone()
74            .acquire_owned()
75            .await
76            .map_err(|_| PasswordWorkError::Closed)?;
77        let password = password.to_owned();
78        let policy = self.policy;
79
80        tokio::task::spawn_blocking(move || {
81            let _permit = permit;
82            hash_password_with_policy(&password, &policy)
83        })
84        .await
85        .map_err(|_| PasswordWorkError::Worker)?
86        .map_err(PasswordWorkError::from)
87    }
88
89    /// Fail immediately when the CPU/memory budget is occupied. The blocking job
90    /// owns the permit, so cancellation cannot release capacity prematurely.
91    ///
92    /// # Errors
93    /// Returns `Busy` on saturation, or a worker/cryptographic error.
94    pub async fn try_verify_password(
95        &self,
96        password: &str,
97        stored_hash: &str,
98    ) -> Result<PasswordHashVerification> {
99        let permit = self
100            .semaphore
101            .clone()
102            .try_acquire_owned()
103            .map_err(|_| PasswordWorkError::Busy)?;
104        self.verify_with_permit(password, stored_hash, permit).await
105    }
106
107    /// Verifies a password on the blocking pool after acquiring an execution permit.
108    ///
109    /// # Errors
110    /// Returns an error when the worker fails or the stored verifier is invalid.
111    pub async fn verify_password(
112        &self,
113        password: &str,
114        stored_hash: &str,
115    ) -> Result<PasswordHashVerification> {
116        let permit = self
117            .semaphore
118            .clone()
119            .acquire_owned()
120            .await
121            .map_err(|_| PasswordWorkError::Closed)?;
122        self.verify_with_permit(password, stored_hash, permit).await
123    }
124
125    async fn verify_with_permit(
126        &self,
127        password: &str,
128        stored_hash: &str,
129        permit: tokio::sync::OwnedSemaphorePermit,
130    ) -> Result<PasswordHashVerification> {
131        let password = password.to_owned();
132        let stored_hash = stored_hash.to_owned();
133        let policy = self.policy;
134
135        tokio::task::spawn_blocking(move || {
136            let _permit = permit;
137            verify_password_with_policy(&password, &stored_hash, &policy)
138        })
139        .await
140        .map_err(|_| PasswordWorkError::Worker)?
141        .map_err(PasswordWorkError::from)
142    }
143}
144
145impl Default for PasswordHashRuntime {
146    fn default() -> Self {
147        Self {
148            policy: PasswordHashPolicy::default(),
149            semaphore: Arc::new(Semaphore::new(DEFAULT_PASSWORD_HASH_MAX_CONCURRENCY)),
150        }
151    }
152}
153
154#[cfg(test)]
155mod tests {
156    use std::time::Duration;
157
158    use aster_forge_crypto::{
159        CryptoError, PasswordHashVerificationLimits, PasswordHashWorkFactor,
160        hash_password_with_policy,
161    };
162
163    use super::*;
164
165    fn policy(memory_kib: u32) -> PasswordHashPolicy {
166        PasswordHashPolicy::new(
167            PasswordHashWorkFactor::new(memory_kib, 1, 1, 32).unwrap(),
168            PasswordHashVerificationLimits::new(memory_kib, 1, 1, 32).unwrap(),
169        )
170        .unwrap()
171    }
172
173    #[tokio::test]
174    async fn immediate_verification_rejects_saturation_and_instances_are_independent() {
175        let runtime = PasswordHashRuntime::with_policy(1, policy(8)).unwrap();
176        let independent = PasswordHashRuntime::with_policy(1, policy(8)).unwrap();
177        let hash = hash_password_with_policy("secret", &policy(8)).unwrap();
178        let permit = runtime.semaphore.clone().acquire_owned().await.unwrap();
179        assert!(matches!(
180            runtime.try_verify_password("secret", &hash).await,
181            Err(PasswordWorkError::Busy)
182        ));
183        assert!(
184            independent
185                .try_verify_password("secret", &hash)
186                .await
187                .unwrap()
188                .is_valid
189        );
190        drop(permit);
191        assert!(
192            runtime
193                .try_verify_password("secret", &hash)
194                .await
195                .unwrap()
196                .is_valid
197        );
198    }
199
200    #[test]
201    fn cancelling_a_request_does_not_release_a_queued_blocking_job_permit() {
202        let executor = tokio::runtime::Builder::new_current_thread()
203            .enable_all()
204            .max_blocking_threads(1)
205            .build()
206            .unwrap();
207        executor.block_on(async {
208            let (entered, wait_entered) = tokio::sync::oneshot::channel();
209            let (release, gate) = std::sync::mpsc::channel();
210            let blocker = tokio::task::spawn_blocking(move || {
211                entered.send(()).unwrap();
212                gate.recv().unwrap();
213            });
214            wait_entered.await.unwrap();
215            let runtime = PasswordHashRuntime::with_policy(1, policy(8)).unwrap();
216            let hash = hash_password_with_policy("secret", &policy(8)).unwrap();
217            let task_runtime = runtime.clone();
218            let task_hash = hash.clone();
219            let task = tokio::spawn(async move {
220                task_runtime.try_verify_password("secret", &task_hash).await
221            });
222            tokio::time::timeout(Duration::from_secs(1), async {
223                while runtime.semaphore.available_permits() != 0 {
224                    tokio::task::yield_now().await;
225                }
226            })
227            .await
228            .unwrap();
229            task.abort();
230            assert!(task.await.unwrap_err().is_cancelled());
231            assert_eq!(runtime.semaphore.available_permits(), 0);
232            assert!(matches!(
233                runtime.try_verify_password("secret", &hash).await,
234                Err(PasswordWorkError::Busy)
235            ));
236            release.send(()).unwrap();
237            blocker.await.unwrap();
238            tokio::time::timeout(Duration::from_secs(1), async {
239                while runtime.semaphore.available_permits() != 1 {
240                    tokio::task::yield_now().await;
241                }
242            })
243            .await
244            .unwrap();
245            assert!(
246                runtime
247                    .try_verify_password("secret", &hash)
248                    .await
249                    .unwrap()
250                    .is_valid
251            );
252        });
253    }
254
255    #[test]
256    fn rejects_zero_concurrency() {
257        let error = PasswordHashRuntime::new(0).unwrap_err();
258        assert!(matches!(error, PasswordWorkError::InvalidConcurrency));
259    }
260
261    #[test]
262    fn production_policy_uses_forge_default_profile() {
263        let runtime = PasswordHashRuntime::new(1).unwrap();
264        let work_factor = runtime.policy.work_factor();
265
266        assert_eq!(work_factor.memory_kib(), 64 * 1024);
267        assert_eq!(work_factor.iterations(), 3);
268        assert_eq!(work_factor.parallelism(), 4);
269        assert_eq!(work_factor.output_length(), 32);
270    }
271
272    #[tokio::test]
273    async fn hash_work_waits_for_the_concurrency_permit() {
274        let runtime = PasswordHashRuntime::with_policy(1, policy(8)).unwrap();
275        let permit = runtime.semaphore.clone().acquire_owned().await.unwrap();
276        let runtime_for_task = runtime.clone();
277        let mut task =
278            tokio::spawn(async move { runtime_for_task.hash_password("bounded-password").await });
279
280        assert!(
281            tokio::time::timeout(Duration::from_millis(20), &mut task)
282                .await
283                .is_err()
284        );
285
286        drop(permit);
287        let hash = tokio::time::timeout(Duration::from_secs(1), task)
288            .await
289            .unwrap()
290            .unwrap()
291            .unwrap();
292        assert!(hash.starts_with("$argon2id$v=19$m=8,t=1,p=1$"));
293    }
294
295    #[tokio::test]
296    async fn verification_reports_legacy_rehash_and_wrong_password() {
297        let current_policy = policy(16);
298        let legacy_policy = policy(8);
299        let runtime = PasswordHashRuntime::with_policy(1, current_policy).unwrap();
300        let legacy_hash = hash_password_with_policy("legacy-password", &legacy_policy).unwrap();
301
302        let verified = runtime
303            .verify_password("legacy-password", &legacy_hash)
304            .await
305            .unwrap();
306        assert!(verified.is_valid);
307        assert!(verified.needs_rehash);
308
309        let rejected = runtime
310            .verify_password("wrong-password", &legacy_hash)
311            .await
312            .unwrap();
313        assert!(!rejected.is_valid);
314        assert!(!rejected.needs_rehash);
315    }
316
317    #[tokio::test]
318    async fn malformed_hash_error_releases_the_concurrency_permit() {
319        let runtime = PasswordHashRuntime::with_policy(1, policy(8)).unwrap();
320        let error = runtime
321            .verify_password("password", "not-a-phc-string")
322            .await
323            .unwrap_err();
324        assert!(matches!(error, PasswordWorkError::Crypto(_)));
325
326        let hash = runtime.hash_password("next-password").await.unwrap();
327        assert!(
328            runtime
329                .verify_password("next-password", &hash)
330                .await
331                .unwrap()
332                .is_valid
333        );
334    }
335
336    #[test]
337    fn over_budget_hash_is_classified_before_verification() {
338        let runtime = PasswordHashRuntime::with_policy(1, policy(8)).unwrap();
339        let hash = hash_password_with_policy("password", &policy(8)).unwrap();
340        let over_budget = hash.replacen("m=8", "m=9", 1);
341        let error =
342            verify_password_with_policy("password", &over_budget, &runtime.policy).unwrap_err();
343
344        assert!(matches!(
345            error,
346            CryptoError::PasswordHashVerificationLimit {
347                parameter: "m",
348                actual: 9,
349                maximum: 8,
350            }
351        ));
352    }
353}
aster_forge_auth/password.rs

aster_forge_auth/
password.rs
