1use std::collections::BTreeMap;
2use std::path::{Path, PathBuf};
3use std::sync::Arc;
4
5use async_trait::async_trait;
6use futures::stream::BoxStream;
7use serde::{Deserialize, Serialize};
8use sha2::{Digest, Sha256};
9use subtle::ConstantTimeEq;
10
11use crate::event_log::{
12 AnyEventLog, CompactReport, ConsumerId, EventId, EventLog, EventLogDescription, LogError,
13 LogEvent, Topic,
14};
15use crate::orchestration::CapabilityPolicy;
16use crate::secrets::{
17 ensure_scoped_secret_access_allowed, RotationHandle, SecretBytes, SecretDeleteRequest,
18 SecretError, SecretId, SecretMeta, SecretProvider,
19};
20use crate::TenantId;
21
22pub const TENANT_REGISTRY_DIR: &str = "tenants";
23pub const TENANT_REGISTRY_FILE: &str = "registry.json";
24pub const TENANT_SECRET_NAMESPACE_PREFIX: &str = "harn.tenant.";
25pub const TENANT_EVENT_TOPIC_PREFIX: &str = "tenant.";
26
27#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
28#[serde(transparent)]
29pub struct ApiKeyId(pub String);
30
31#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
32#[serde(default)]
33pub struct TenantBudget {
34 pub daily_cost_usd: Option<f64>,
35 pub hourly_cost_usd: Option<f64>,
36 pub ingest_per_minute: Option<u32>,
37 pub event_log_size_bytes: u64,
38 pub in_flight_dispatches: u32,
39 pub dlq_entries: u32,
40}
41
42impl Default for TenantBudget {
43 fn default() -> Self {
44 Self {
45 daily_cost_usd: None,
46 hourly_cost_usd: None,
47 ingest_per_minute: None,
48 event_log_size_bytes: 10 * 1024 * 1024 * 1024,
49 in_flight_dispatches: 100,
50 dlq_entries: 10_000,
51 }
52 }
53}
54
55#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
56pub struct TenantScope {
57 pub id: TenantId,
58 pub state_root: PathBuf,
59 pub secret_namespace: String,
60 pub event_log_topic_prefix: String,
61 pub capability_ceiling: CapabilityPolicy,
62 pub budget: TenantBudget,
63 pub api_key_ids: Vec<ApiKeyId>,
64}
65
66impl TenantScope {
67 pub fn new(id: TenantId, orchestrator_state_root: impl AsRef<Path>) -> Result<Self, String> {
68 validate_tenant_id(&id.0)?;
69 let state_root = orchestrator_state_root
70 .as_ref()
71 .join(TENANT_REGISTRY_DIR)
72 .join(&id.0);
73 Ok(Self {
74 secret_namespace: tenant_secret_namespace(&id),
75 event_log_topic_prefix: tenant_event_topic_prefix(&id),
76 id,
77 state_root,
78 capability_ceiling: CapabilityPolicy::default(),
79 budget: TenantBudget::default(),
80 api_key_ids: Vec::new(),
81 })
82 }
83
84 pub fn topic(&self, topic: &Topic) -> Result<Topic, LogError> {
85 tenant_topic(&self.id, topic)
86 }
87}
88
89#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
90#[serde(rename_all = "snake_case")]
91pub enum TenantStatus {
92 Active,
93 Suspended,
94}
95
96#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
97pub struct TenantApiKeyRecord {
98 pub id: ApiKeyId,
99 pub hash_sha256: String,
100 pub prefix: String,
101 pub created_at: String,
102}
103
104#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
105pub struct TenantRecord {
106 pub scope: TenantScope,
107 pub status: TenantStatus,
108 pub created_at: String,
109 pub suspended_at: Option<String>,
110 pub api_keys: Vec<TenantApiKeyRecord>,
111}
112
113#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
114#[serde(default)]
115pub struct TenantRegistrySnapshot {
116 pub tenants: Vec<TenantRecord>,
117}
118
119#[derive(Clone, Debug)]
120pub struct TenantStore {
121 state_dir: PathBuf,
122 tenants: BTreeMap<String, TenantRecord>,
123}
124
125#[derive(Clone, Debug, PartialEq, Eq)]
126pub enum TenantResolutionError {
127 Unknown,
128 Suspended(TenantId),
129}
130
131impl TenantStore {
132 pub fn load(state_dir: impl AsRef<Path>) -> Result<Self, String> {
133 let state_dir = state_dir.as_ref().to_path_buf();
134 let path = registry_path(&state_dir);
135 if !path.is_file() {
136 return Ok(Self {
137 state_dir,
138 tenants: BTreeMap::new(),
139 });
140 }
141 let content = std::fs::read_to_string(&path)
142 .map_err(|error| format!("failed to read {}: {error}", path.display()))?;
143 let snapshot: TenantRegistrySnapshot = serde_json::from_str(&content)
144 .map_err(|error| format!("failed to parse {}: {error}", path.display()))?;
145 let tenants = snapshot
146 .tenants
147 .into_iter()
148 .map(|record| (record.scope.id.0.clone(), record))
149 .collect();
150 Ok(Self { state_dir, tenants })
151 }
152
153 pub fn save(&self) -> Result<(), String> {
154 let dir = self.state_dir.join(TENANT_REGISTRY_DIR);
155 std::fs::create_dir_all(&dir)
156 .map_err(|error| format!("failed to create {}: {error}", dir.display()))?;
157 let snapshot = TenantRegistrySnapshot {
158 tenants: self.list(),
159 };
160 let encoded = serde_json::to_string_pretty(&snapshot).map_err(|error| error.to_string())?;
161 let path = registry_path(&self.state_dir);
162 write_file_replace(&path, encoded.as_bytes())
163 .map_err(|error| format!("failed to write {}: {error}", path.display()))
164 }
165
166 pub fn create_tenant(
167 &mut self,
168 id: impl Into<String>,
169 budget: TenantBudget,
170 ) -> Result<(TenantRecord, String), String> {
171 let id = id.into();
172 validate_tenant_id(&id)?;
173 if self.tenants.contains_key(&id) {
174 return Err(format!("tenant '{id}' already exists"));
175 }
176 let api_key = generate_api_key(&id);
177 let api_key_id = ApiKeyId(format!("key_{}", uuid::Uuid::now_v7()));
178 let created_at = now_rfc3339();
179 let mut scope = TenantScope::new(TenantId::new(id.clone()), &self.state_dir)?;
180 scope.budget = budget;
181 scope.api_key_ids.push(api_key_id.clone());
182 std::fs::create_dir_all(&scope.state_root).map_err(|error| {
183 format!(
184 "failed to create tenant state dir {}: {error}",
185 scope.state_root.display()
186 )
187 })?;
188 let record = TenantRecord {
189 scope,
190 status: TenantStatus::Active,
191 created_at: created_at.clone(),
192 suspended_at: None,
193 api_keys: vec![TenantApiKeyRecord {
194 id: api_key_id,
195 hash_sha256: api_key_hash(&api_key),
196 prefix: api_key_prefix(&api_key),
197 created_at,
198 }],
199 };
200 self.tenants.insert(id, record.clone());
201 self.save()?;
202 Ok((record, api_key))
203 }
204
205 pub fn list(&self) -> Vec<TenantRecord> {
206 self.tenants.values().cloned().collect()
207 }
208
209 pub fn get(&self, id: &str) -> Option<&TenantRecord> {
210 self.tenants.get(id)
211 }
212
213 pub fn resolve_id(&self, id: &str) -> Result<TenantScope, TenantResolutionError> {
214 self.tenants
215 .get(id)
216 .ok_or(TenantResolutionError::Unknown)
217 .and_then(active_tenant_scope)
218 }
219
220 pub fn suspend(&mut self, id: &str) -> Result<TenantRecord, String> {
221 let record = self
222 .tenants
223 .get_mut(id)
224 .ok_or_else(|| format!("unknown tenant '{id}'"))?;
225 record.status = TenantStatus::Suspended;
226 record.suspended_at = Some(now_rfc3339());
227 let record = record.clone();
228 self.save()?;
229 Ok(record)
230 }
231
232 pub fn delete(&mut self, id: &str) -> Result<TenantRecord, String> {
233 let record = self
234 .tenants
235 .remove(id)
236 .ok_or_else(|| format!("unknown tenant '{id}'"))?;
237 if record.scope.state_root.exists() {
238 std::fs::remove_dir_all(&record.scope.state_root).map_err(|error| {
239 format!(
240 "failed to remove tenant state dir {}: {error}",
241 record.scope.state_root.display()
242 )
243 })?;
244 }
245 self.save()?;
246 Ok(record)
247 }
248
249 pub fn resolve_api_key(&self, candidate: &str) -> Result<TenantScope, TenantResolutionError> {
250 let candidate_hash = api_key_hash(candidate);
251 for record in self.tenants.values() {
252 let matched = record.api_keys.iter().any(|key| {
253 key.hash_sha256
254 .as_bytes()
255 .ct_eq(candidate_hash.as_bytes())
256 .into()
257 });
258 if matched {
259 return active_tenant_scope(record);
260 }
261 }
262 Err(TenantResolutionError::Unknown)
263 }
264}
265
266fn active_tenant_scope(record: &TenantRecord) -> Result<TenantScope, TenantResolutionError> {
267 match record.status {
268 TenantStatus::Active => Ok(record.scope.clone()),
269 TenantStatus::Suspended => Err(TenantResolutionError::Suspended(record.scope.id.clone())),
270 }
271}
272
273pub struct TenantEventLog {
274 inner: Arc<AnyEventLog>,
275 scope: TenantScope,
276}
277
278impl TenantEventLog {
279 pub fn new(inner: Arc<AnyEventLog>, scope: TenantScope) -> Self {
280 Self { inner, scope }
281 }
282
283 pub fn scope(&self) -> &TenantScope {
284 &self.scope
285 }
286
287 fn scoped_topic(&self, topic: &Topic) -> Result<Topic, LogError> {
288 if topic.as_str().starts_with(TENANT_EVENT_TOPIC_PREFIX) {
289 if topic
290 .as_str()
291 .starts_with(&self.scope.event_log_topic_prefix)
292 {
293 return Ok(topic.clone());
294 }
295 return Err(LogError::InvalidTopic(format!(
296 "topic '{}' is outside tenant scope '{}'",
297 topic.as_str(),
298 self.scope.id.0
299 )));
300 }
301 self.scope.topic(topic)
302 }
303}
304
305impl EventLog for TenantEventLog {
306 fn describe(&self) -> EventLogDescription {
307 self.inner.describe()
308 }
309
310 async fn append(&self, topic: &Topic, event: LogEvent) -> Result<EventId, LogError> {
311 self.inner.append(&self.scoped_topic(topic)?, event).await
312 }
313
314 async fn flush(&self) -> Result<(), LogError> {
315 self.inner.flush().await
316 }
317
318 async fn read_range(
319 &self,
320 topic: &Topic,
321 from: Option<EventId>,
322 limit: usize,
323 ) -> Result<Vec<(EventId, LogEvent)>, LogError> {
324 self.inner
325 .read_range(&self.scoped_topic(topic)?, from, limit)
326 .await
327 }
328
329 async fn subscribe(
330 self: Arc<Self>,
331 topic: &Topic,
332 from: Option<EventId>,
333 ) -> Result<BoxStream<'static, Result<(EventId, LogEvent), LogError>>, LogError> {
334 self.inner
335 .clone()
336 .subscribe(&self.scoped_topic(topic)?, from)
337 .await
338 }
339
340 async fn ack(
341 &self,
342 topic: &Topic,
343 consumer: &ConsumerId,
344 up_to: EventId,
345 ) -> Result<(), LogError> {
346 self.inner
347 .ack(&self.scoped_topic(topic)?, consumer, up_to)
348 .await
349 }
350
351 async fn consumer_cursor(
352 &self,
353 topic: &Topic,
354 consumer: &ConsumerId,
355 ) -> Result<Option<EventId>, LogError> {
356 self.inner
357 .consumer_cursor(&self.scoped_topic(topic)?, consumer)
358 .await
359 }
360
361 async fn latest(&self, topic: &Topic) -> Result<Option<EventId>, LogError> {
362 self.inner.latest(&self.scoped_topic(topic)?).await
363 }
364
365 async fn compact(&self, topic: &Topic, before: EventId) -> Result<CompactReport, LogError> {
366 self.inner.compact(&self.scoped_topic(topic)?, before).await
367 }
368}
369
370pub struct TenantSecretProvider {
371 inner: Arc<dyn SecretProvider>,
372 scope: TenantScope,
373}
374
375impl TenantSecretProvider {
376 pub fn new(inner: Arc<dyn SecretProvider>, scope: TenantScope) -> Self {
377 Self { inner, scope }
378 }
379
380 fn scoped_id(&self, operation: &str, id: &SecretId) -> Result<SecretId, SecretError> {
381 let own_prefix = format!("{}.", self.scope.secret_namespace);
382 if id.namespace == self.scope.secret_namespace || id.namespace.starts_with(&own_prefix) {
383 return Ok(id.clone());
384 }
385 if id.namespace.starts_with(TENANT_SECRET_NAMESPACE_PREFIX) {
386 return Err(SecretError::AccessDenied {
387 operation: operation.to_string(),
388 id: id.clone(),
389 message: format!("namespace is outside tenant '{}'", self.scope.id.0),
390 });
391 }
392 Ok(SecretId {
393 namespace: format!("{}.{}", self.scope.secret_namespace, id.namespace),
394 name: id.name.clone(),
395 version: id.version.clone(),
396 })
397 }
398}
399
400#[async_trait]
401impl SecretProvider for TenantSecretProvider {
402 async fn get(&self, id: &SecretId) -> Result<SecretBytes, SecretError> {
403 self.inner.get(&self.scoped_id("read", id)?).await
404 }
405
406 async fn put(&self, id: &SecretId, value: SecretBytes) -> Result<(), SecretError> {
407 self.inner.put(&self.scoped_id("write", id)?, value).await
408 }
409
410 async fn rotate(&self, id: &SecretId) -> Result<RotationHandle, SecretError> {
411 self.inner.rotate(&self.scoped_id("rotate", id)?).await
412 }
413
414 async fn list(&self, prefix: &SecretId) -> Result<Vec<SecretMeta>, SecretError> {
415 self.inner.list(&self.scoped_id("list", prefix)?).await
416 }
417
418 async fn delete_scoped(&self, request: SecretDeleteRequest) -> Result<(), SecretError> {
419 ensure_scoped_secret_access_allowed("delete", &request.id)?;
420 self.inner
421 .delete_scoped(SecretDeleteRequest {
422 id: self.scoped_id("delete", &request.id)?,
423 scope: request.scope,
424 audit: request.audit,
425 })
426 .await
427 }
428
429 fn namespace(&self) -> &str {
430 &self.scope.secret_namespace
431 }
432
433 fn supports_versions(&self) -> bool {
434 self.inner.supports_versions()
435 }
436}
437
438pub fn tenant_event_topic_prefix(id: &TenantId) -> String {
439 format!("{TENANT_EVENT_TOPIC_PREFIX}{}.", id.0)
440}
441
442pub fn tenant_secret_namespace(id: &TenantId) -> String {
443 format!("{TENANT_SECRET_NAMESPACE_PREFIX}{}", id.0)
444}
445
446pub fn tenant_topic(id: &TenantId, topic: &Topic) -> Result<Topic, LogError> {
447 validate_tenant_id(&id.0).map_err(LogError::InvalidTopic)?;
448 let prefix = tenant_event_topic_prefix(id);
449 if topic.as_str().starts_with(&prefix) {
450 return Ok(topic.clone());
451 }
452 if topic.as_str().starts_with(TENANT_EVENT_TOPIC_PREFIX) {
453 return Err(LogError::InvalidTopic(format!(
454 "topic '{}' is outside tenant scope '{}'",
455 topic.as_str(),
456 id.0
457 )));
458 }
459 Topic::new(format!("{prefix}{}", topic.as_str()))
460}
461
462pub fn validate_tenant_id(id: &str) -> Result<(), String> {
463 if id.trim().is_empty() {
464 return Err("tenant id cannot be empty".to_string());
465 }
466 if !id
467 .chars()
468 .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '_' | '-'))
469 {
470 return Err(format!(
471 "tenant id '{id}' contains unsupported characters; use ASCII letters, numbers, '_' or '-'"
472 ));
473 }
474 Ok(())
475}
476
477fn registry_path(state_dir: &Path) -> PathBuf {
478 state_dir
479 .join(TENANT_REGISTRY_DIR)
480 .join(TENANT_REGISTRY_FILE)
481}
482
483fn write_file_replace(path: &Path, contents: &[u8]) -> std::io::Result<()> {
484 crate::atomic_io::atomic_write(path, contents)
485}
486
487fn generate_api_key(id: &str) -> String {
488 let random: [u8; 32] = rand::random();
489 format!("harn_tenant_{id}_{}", hex::encode(random))
490}
491
492fn api_key_hash(value: &str) -> String {
493 hex::encode(Sha256::digest(value.as_bytes()))
494}
495
496fn api_key_prefix(value: &str) -> String {
497 value.chars().take(18).collect()
498}
499
500fn now_rfc3339() -> String {
501 harn_clock::system_now_rfc3339()
502}
503
504#[cfg(test)]
505mod tests {
506 use std::collections::BTreeMap;
507 use std::sync::Mutex;
508
509 use async_trait::async_trait;
510
511 use super::*;
512 use crate::event_log::{EventLog, MemoryEventLog};
513
514 #[tokio::test]
515 async fn tenant_event_log_enforces_topic_prefix() {
516 let inner = Arc::new(AnyEventLog::Memory(MemoryEventLog::new(8)));
517 let scope =
518 TenantScope::new(TenantId::new("tenant-a"), std::env::temp_dir()).expect("scope");
519 let tenant_log = Arc::new(TenantEventLog::new(inner.clone(), scope));
520 let base = Topic::new("trigger.outbox").unwrap();
521
522 tenant_log
523 .append(&base, LogEvent::new("ok", serde_json::json!({"n": 1})))
524 .await
525 .unwrap();
526
527 let scoped = Topic::new("tenant.tenant-a.trigger.outbox").unwrap();
528 assert_eq!(inner.read_range(&scoped, None, 10).await.unwrap().len(), 1);
529 let other = Topic::new("tenant.tenant-b.trigger.outbox").unwrap();
530 assert!(tenant_log
531 .append(&other, LogEvent::new("bad", serde_json::json!({})))
532 .await
533 .is_err());
534 }
535
536 struct MemorySecretProvider {
537 namespace: String,
538 values: Mutex<BTreeMap<SecretId, SecretBytes>>,
539 }
540
541 #[async_trait]
542 impl SecretProvider for MemorySecretProvider {
543 async fn get(&self, id: &SecretId) -> Result<SecretBytes, SecretError> {
544 self.values
545 .lock()
546 .expect("secret map")
547 .get(id)
548 .map(SecretBytes::reborrow)
549 .ok_or_else(|| SecretError::NotFound {
550 provider: self.namespace.clone(),
551 id: id.clone(),
552 })
553 }
554
555 async fn put(&self, id: &SecretId, value: SecretBytes) -> Result<(), SecretError> {
556 self.values
557 .lock()
558 .expect("secret map")
559 .insert(id.clone(), value);
560 Ok(())
561 }
562
563 async fn rotate(&self, _id: &SecretId) -> Result<RotationHandle, SecretError> {
564 Err(SecretError::Unsupported {
565 provider: self.namespace.clone(),
566 operation: "rotate",
567 })
568 }
569
570 async fn list(&self, _prefix: &SecretId) -> Result<Vec<SecretMeta>, SecretError> {
571 Ok(Vec::new())
572 }
573
574 fn namespace(&self) -> &str {
575 &self.namespace
576 }
577
578 fn supports_versions(&self) -> bool {
579 false
580 }
581 }
582
583 #[tokio::test]
584 async fn tenant_secret_provider_rescopes_and_denies_cross_tenant_ids() {
585 let inner = Arc::new(MemorySecretProvider {
586 namespace: "global".to_string(),
587 values: Mutex::new(BTreeMap::new()),
588 });
589 let scope =
590 TenantScope::new(TenantId::new("tenant-a"), std::env::temp_dir()).expect("scope");
591 let provider = TenantSecretProvider::new(inner.clone(), scope.clone());
592
593 provider
594 .put(
595 &SecretId::new("github", "webhook"),
596 SecretBytes::from("a-secret"),
597 )
598 .await
599 .unwrap();
600
601 let scoped_id = SecretId::new(format!("{}.github", scope.secret_namespace), "webhook");
602 let value = inner.get(&scoped_id).await.unwrap();
603 value.with_exposed(|bytes| assert_eq!(bytes, b"a-secret"));
604
605 provider
606 .put(
607 &SecretId::new("slack", "webhook"),
608 SecretBytes::from("slack-secret"),
609 )
610 .await
611 .unwrap();
612 let slack_id = SecretId::new(format!("{}.slack", scope.secret_namespace), "webhook");
613 let slack = inner.get(&slack_id).await.unwrap();
614 slack.with_exposed(|bytes| assert_eq!(bytes, b"slack-secret"));
615
616 let cross = SecretId::new("harn.tenant.tenant-b.github", "webhook");
617 assert!(matches!(
618 provider.get(&cross).await,
619 Err(SecretError::AccessDenied { operation, .. }) if operation == "read"
620 ));
621 }
622
623 #[test]
624 fn tenant_store_save_replaces_registry_without_temp_leak() {
625 let temp = tempfile::tempdir().unwrap();
626 let mut store = TenantStore::load(temp.path()).unwrap();
627 store
628 .create_tenant("tenant-a", TenantBudget::default())
629 .unwrap();
630
631 let registry = registry_path(temp.path());
632 assert!(registry.is_file());
633 let leaked_temp = std::fs::read_dir(registry.parent().unwrap())
634 .unwrap()
635 .filter_map(Result::ok)
636 .any(|entry| entry.file_name().to_string_lossy().ends_with(".tmp"));
637 assert!(!leaked_temp);
638 }
639}