1use std::{
2 collections::HashMap,
3 path::Path,
4 sync::{Arc, Mutex, MutexGuard},
5};
6
7use kcode_k1_access_format::{AccessAction, decode, encode};
8use kcode_k1_access_projection::{ApplyOutcome, Projection};
9use kcode_k1_peering::K1Peering;
10use kcode_k1_txn_ordering::{K1TxnOrdering, Subsystem as K1Subsystem};
11
12pub use kcode_k1_access_projection::{
13 AccessCheck, AccessId, AccessRevision, Authorizations, GroupId, OwnerWitness, RequestPrincipal,
14 SubsystemId, Target, TxId, UserId,
15};
16
17const SUBSYSTEM_NAME: &str = "k1-access-subsystem";
18type OperationId = [u8; 16];
19
20struct Pending {
21 action: AccessAction,
22 result: Option<(TxId, ApplyOutcome)>,
23}
24
25struct DriverState {
26 available: bool,
27 pending: HashMap<OperationId, Pending>,
28}
29
30struct SharedState {
31 inner: Mutex<DriverState>,
32}
33
34impl SharedState {
35 fn new() -> Self {
36 Self {
37 inner: Mutex::new(DriverState {
38 available: true,
39 pending: HashMap::new(),
40 }),
41 }
42 }
43
44 fn lock(&self) -> Result<MutexGuard<'_, DriverState>, String> {
45 self.inner
46 .lock()
47 .map_err(|_| "k1 access driver state lock failed".to_owned())
48 }
49
50 fn ensure_available(&self) -> Result<(), String> {
51 if self.lock()?.available {
52 Ok(())
53 } else {
54 Err("k1 access driver unavailable".to_owned())
55 }
56 }
57
58 fn reserve(&self, action: AccessAction) -> Result<OperationId, String> {
59 self.ensure_available()?;
60 loop {
61 let mut operation_id = [0_u8; 16];
62 getrandom::fill(&mut operation_id)
63 .map_err(|error| format!("operation ID randomness failed: {error}"))?;
64 let mut state = self.lock()?;
65 if !state.available {
66 return Err("k1 access driver unavailable".to_owned());
67 }
68 if state.pending.contains_key(&operation_id) {
69 continue;
70 }
71 state
72 .pending
73 .try_reserve(1)
74 .map_err(|_| "pending operation allocation failed".to_owned())?;
75 state.pending.insert(
76 operation_id,
77 Pending {
78 action,
79 result: None,
80 },
81 );
82 return Ok(operation_id);
83 }
84 }
85
86 fn cancel(&self, operation_id: OperationId) -> Result<(), String> {
87 let mut state = self.lock()?;
88 if state.pending.remove(&operation_id).is_none() {
89 invalidate(&mut state);
90 return Err("pending operation disappeared".to_owned());
91 }
92 if state.available {
93 Ok(())
94 } else {
95 Err("k1 access driver unavailable".to_owned())
96 }
97 }
98
99 fn record(
100 &self,
101 operation_id: OperationId,
102 action: &AccessAction,
103 txid: TxId,
104 outcome: ApplyOutcome,
105 ) -> Result<(), String> {
106 let mut state = self.lock()?;
107 if !state.available {
108 return Err("k1 access driver unavailable".to_owned());
109 }
110 let contradiction = match state.pending.get_mut(&operation_id) {
111 None => return Ok(()),
112 Some(pending) if &pending.action != action => {
113 Some("operation ID correlated with a different access action")
114 }
115 Some(pending) if pending.result.is_some() => {
116 Some("duplicate callback evidence for one operation ID")
117 }
118 Some(pending) => {
119 pending.result = Some((txid, outcome));
120 None
121 }
122 };
123 if let Some(error) = contradiction {
124 invalidate(&mut state);
125 Err(error.to_owned())
126 } else {
127 Ok(())
128 }
129 }
130
131 fn finish(
132 &self,
133 operation_id: OperationId,
134 submission: Result<TxId, String>,
135 ) -> Result<AccessRevision, String> {
136 let result = {
137 let mut state = self.lock()?;
138 let Some(pending) = state.pending.remove(&operation_id) else {
139 invalidate(&mut state);
140 return Err("pending operation disappeared".to_owned());
141 };
142 if !state.available {
143 return Err("k1 access driver unavailable".to_owned());
144 }
145 pending.result
146 };
147 match submission {
148 Ok(submitted) => match result {
149 Some((callback, outcome)) if callback == submitted => outcome_result(outcome),
150 Some(_) => {
151 self.fault("submission transaction ID did not match callback transaction ID")
152 }
153 None => self.fault("successful submission had no synchronous callback"),
154 },
155 Err(error) => match result {
156 Some((_, outcome)) => outcome_result(outcome),
157 None => Err(error),
158 },
159 }
160 }
161
162 fn make_unavailable(&self) -> Result<(), String> {
163 let mut state = self.lock()?;
164 invalidate(&mut state);
165 Ok(())
166 }
167
168 fn fault<T>(&self, error: &str) -> Result<T, String> {
169 let _ = self.make_unavailable();
170 Err(error.to_owned())
171 }
172}
173
174fn invalidate(state: &mut DriverState) {
175 state.available = false;
176 for pending in state.pending.values_mut() {
177 pending.result = None;
178 }
179}
180
181fn outcome_result(outcome: ApplyOutcome) -> Result<AccessRevision, String> {
182 match outcome {
183 ApplyOutcome::Applied(revision) | ApplyOutcome::Unchanged(revision) => Ok(revision),
184 ApplyOutcome::Rejected(reason) => Err(reason),
185 }
186}
187
188struct AccessSubsystem {
189 projection: Arc<Projection>,
190 shared: Arc<SharedState>,
191}
192
193impl K1Subsystem for AccessSubsystem {
194 fn submit_txn(&self, id: TxId, payload: &[u8]) -> Result<(), String> {
195 let (operation_id, action) = match decode(payload) {
196 Ok(decoded) => decoded,
197 Err(error) => {
198 let _ = self.shared.make_unavailable();
199 return Err(error);
200 }
201 };
202 self.shared.ensure_available()?;
203 let outcome = match self.projection.apply(id, action.clone()) {
204 Ok(outcome) => outcome,
205 Err(error) => {
206 let _ = self.shared.make_unavailable();
207 return Err(error);
208 }
209 };
210 self.shared.record(operation_id, &action, id, outcome)
211 }
212
213 fn reorg(&self) -> Result<(), String> {
214 let state_result = self.shared.make_unavailable();
215 match self.projection.clear() {
216 Ok(()) => state_result,
217 Err(error) => Err(error),
218 }
219 }
220}
221
222pub struct K1AccessDriver {
223 projection: Arc<Projection>,
224 _ordering: Arc<K1TxnOrdering>,
225 peering: Arc<K1Peering>,
226 shared: Arc<SharedState>,
227 subsystem: SubsystemId,
228}
229
230impl K1AccessDriver {
231 pub fn open(
232 root: &Path,
233 ordering: Arc<K1TxnOrdering>,
234 peering: Arc<K1Peering>,
235 ) -> Result<Self, String> {
236 let (projection, cursor) = Projection::open(root, &ordering)?;
237 let projection = Arc::new(projection);
238 let shared = Arc::new(SharedState::new());
239 let subsystem = SubsystemId::from_str(SUBSYSTEM_NAME)?;
240 let callback = Arc::new(AccessSubsystem {
241 projection: projection.clone(),
242 shared: shared.clone(),
243 });
244 if let Err(error) = ordering.register_subsystem(subsystem, cursor, callback) {
245 let _ = shared.make_unavailable();
246 return Err(error);
247 }
248 Ok(Self {
249 projection,
250 _ordering: ordering,
251 peering,
252 shared,
253 subsystem,
254 })
255 }
256
257 pub fn create(
258 &self,
259 target: Target,
260 authorizations: Authorizations,
261 ) -> Result<AccessRevision, String> {
262 self.mutate(AccessAction::Create {
263 target,
264 authorizations,
265 })
266 }
267
268 pub fn replace(
269 &self,
270 access_id: AccessId,
271 actor: UserId,
272 groups_revision: Option<TxId>,
273 witness: OwnerWitness,
274 authorizations: Authorizations,
275 ) -> Result<AccessRevision, String> {
276 self.mutate(AccessAction::Replace {
277 access_id,
278 actor,
279 groups_revision,
280 witness,
281 authorizations,
282 })
283 }
284
285 pub fn ensure_discovery(&self, access_id: AccessId) -> Result<AccessRevision, String> {
286 self.mutate(AccessAction::EnsureDiscovery { access_id })
287 }
288
289 pub fn discovery_missing(
290 &self,
291 access_id: AccessId,
292 expected_subsystem: SubsystemId,
293 ) -> Result<bool, String> {
294 self.projection_call(
295 self.projection
296 .discovery_missing(access_id, expected_subsystem),
297 )
298 }
299
300 pub fn discovered_for_user(
301 &self,
302 user: UserId,
303 expected_subsystem: SubsystemId,
304 ) -> Result<Vec<AccessId>, String> {
305 self.projection_call(
306 self.projection
307 .discovered_for_user(user, expected_subsystem),
308 )
309 }
310
311 pub fn discovered_for_group(
312 &self,
313 group: GroupId,
314 expected_subsystem: SubsystemId,
315 ) -> Result<Vec<AccessId>, String> {
316 self.projection_call(
317 self.projection
318 .discovered_for_group(group, expected_subsystem),
319 )
320 }
321
322 pub fn owner_witness(
323 &self,
324 access_id: AccessId,
325 user: UserId,
326 user_groups: &[GroupId],
327 ) -> Result<Option<OwnerWitness>, String> {
328 self.projection_call(self.projection.owner_witness(access_id, user, user_groups))
329 }
330
331 pub fn check(
332 &self,
333 principal: RequestPrincipal,
334 access_id: AccessId,
335 expected_subsystem: SubsystemId,
336 user_groups: &[GroupId],
337 model_groups: &[GroupId],
338 groups_revision: Option<TxId>,
339 ) -> Result<AccessCheck, String> {
340 self.projection_call(self.projection.check(
341 principal,
342 access_id,
343 expected_subsystem,
344 user_groups,
345 model_groups,
346 groups_revision,
347 ))
348 }
349
350 fn mutate(&self, action: AccessAction) -> Result<AccessRevision, String> {
351 let operation_id = self.shared.reserve(action.clone())?;
352 let payload = match encode(operation_id, &action) {
353 Ok(payload) => payload,
354 Err(error) => {
355 self.shared.cancel(operation_id)?;
356 return Err(error);
357 }
358 };
359 if let Err(error) = self.shared.ensure_available() {
360 let _ = self.shared.cancel(operation_id);
361 return Err(error);
362 }
363 let submission = self.peering.submit_txn(self.subsystem, &payload);
364 self.shared.finish(operation_id, submission)
365 }
366
367 fn projection_call<T>(&self, result: Result<T, String>) -> Result<T, String> {
368 match result {
369 Ok(value) => {
370 self.shared.ensure_available()?;
371 Ok(value)
372 }
373 Err(error) => {
374 let _ = self.shared.make_unavailable();
375 Err(error)
376 }
377 }
378 }
379}
380
381#[cfg(test)]
382mod tests {
383 use super::*;
384
385 fn tx(byte: u8) -> TxId {
386 TxId::from_bytes([byte; 12])
387 }
388
389 fn action(byte: u8) -> AccessAction {
390 let owner = UserId::from_tx_id(tx(byte));
391 AccessAction::Create {
392 target: Target::new(SubsystemId::from_str("test").unwrap(), vec![byte]),
393 authorizations: Authorizations::new(
394 vec![kcode_k1_access_projection::OwnerSubject::User(owner)],
395 Vec::new(),
396 )
397 .unwrap(),
398 }
399 }
400
401 fn pending(shared: &SharedState, operation_id: OperationId, action: AccessAction) {
402 shared.lock().unwrap().pending.insert(
403 operation_id,
404 Pending {
405 action,
406 result: None,
407 },
408 );
409 }
410
411 #[test]
412 fn recorded_callback_controls_submission_resolution() {
413 let shared = SharedState::new();
414 let revision = AccessRevision::new(AccessId::new(tx(2)), tx(3));
415 let action = action(1);
416 pending(&shared, [1; 16], action.clone());
417 shared
418 .record([1; 16], &action, tx(4), ApplyOutcome::Applied(revision))
419 .unwrap();
420 assert_eq!(shared.finish([1; 16], Ok(tx(4))), Ok(revision));
421
422 pending(&shared, [2; 16], action.clone());
423 shared
424 .record(
425 [2; 16],
426 &action,
427 tx(5),
428 ApplyOutcome::Rejected("exact rejection".to_owned()),
429 )
430 .unwrap();
431 assert_eq!(
432 shared.finish([2; 16], Err("committed".to_owned())),
433 Err("exact rejection".to_owned())
434 );
435 }
436
437 #[test]
438 fn missing_mismatched_and_duplicate_evidence_fault() {
439 let missing = SharedState::new();
440 pending(&missing, [1; 16], action(1));
441 assert!(missing.finish([1; 16], Ok(tx(1))).is_err());
442 assert!(missing.ensure_available().is_err());
443
444 let mismatched = SharedState::new();
445 pending(&mismatched, [2; 16], action(2));
446 assert!(
447 mismatched
448 .record(
449 [2; 16],
450 &action(3),
451 tx(2),
452 ApplyOutcome::Rejected("remote".to_owned()),
453 )
454 .is_err()
455 );
456 assert!(mismatched.ensure_available().is_err());
457
458 let duplicate = SharedState::new();
459 let exact = action(4);
460 let revision = AccessRevision::new(AccessId::new(tx(4)), tx(5));
461 pending(&duplicate, [3; 16], exact.clone());
462 duplicate
463 .record([3; 16], &exact, tx(5), ApplyOutcome::Applied(revision))
464 .unwrap();
465 assert!(
466 duplicate
467 .record([3; 16], &exact, tx(5), ApplyOutcome::Applied(revision),)
468 .is_err()
469 );
470 assert!(duplicate.ensure_available().is_err());
471 }
472
473 #[test]
474 fn ensure_uses_the_exact_action() {
475 let access_id = AccessId::new(tx(6));
476 let action = AccessAction::EnsureDiscovery { access_id };
477 let shared = SharedState::new();
478 let revision = AccessRevision::new(access_id, tx(5));
479 pending(&shared, [6; 16], action.clone());
480 shared
481 .record([6; 16], &action, tx(7), ApplyOutcome::Unchanged(revision))
482 .unwrap();
483 assert_eq!(shared.finish([6; 16], Ok(tx(7))), Ok(revision));
484 }
485}