1use std::sync::Arc;
12
13use async_trait::async_trait;
14use http::{Method, StatusCode};
15use parking_lot::RwLockWriteGuard;
16use percent_encoding::percent_decode_str;
17use serde_json::{json, Map, Value};
18use tokio::sync::Mutex as AsyncMutex;
19use uuid::Uuid;
20
21use fakecloud_core::multi_account::MultiAccountState;
22use fakecloud_core::service::{AwsRequest, AwsResponse, AwsService, AwsServiceError};
23use fakecloud_persistence::SnapshotStore;
24
25use crate::persistence::save_snapshot;
26use crate::state::{EfsData, SharedEfsState};
27
28pub const EFS_ACTIONS: &[&str] = &[
30 "CreateAccessPoint",
31 "CreateFileSystem",
32 "CreateMountTarget",
33 "CreateReplicationConfiguration",
34 "CreateTags",
35 "DeleteAccessPoint",
36 "DeleteFileSystem",
37 "DeleteFileSystemPolicy",
38 "DeleteMountTarget",
39 "DeleteReplicationConfiguration",
40 "DeleteTags",
41 "DescribeAccessPoints",
42 "DescribeAccountPreferences",
43 "DescribeBackupPolicy",
44 "DescribeFileSystemPolicy",
45 "DescribeFileSystems",
46 "DescribeLifecycleConfiguration",
47 "DescribeMountTargetSecurityGroups",
48 "DescribeMountTargets",
49 "DescribeReplicationConfigurations",
50 "DescribeTags",
51 "ListTagsForResource",
52 "ModifyMountTargetSecurityGroups",
53 "PutAccountPreferences",
54 "PutBackupPolicy",
55 "PutFileSystemPolicy",
56 "PutLifecycleConfiguration",
57 "TagResource",
58 "UntagResource",
59 "UpdateFileSystem",
60 "UpdateFileSystemProtection",
61];
62
63const MUTATING: &[&str] = &[
65 "CreateAccessPoint",
66 "CreateFileSystem",
67 "CreateMountTarget",
68 "CreateReplicationConfiguration",
69 "CreateTags",
70 "DeleteAccessPoint",
71 "DeleteFileSystem",
72 "DeleteFileSystemPolicy",
73 "DeleteMountTarget",
74 "DeleteReplicationConfiguration",
75 "DeleteTags",
76 "ModifyMountTargetSecurityGroups",
77 "PutAccountPreferences",
78 "PutBackupPolicy",
79 "PutFileSystemPolicy",
80 "PutLifecycleConfiguration",
81 "TagResource",
82 "UntagResource",
83 "UpdateFileSystem",
84 "UpdateFileSystemProtection",
85];
86
87pub struct EfsService {
88 state: SharedEfsState,
89 snapshot_store: Option<Arc<dyn SnapshotStore>>,
90 snapshot_lock: Arc<AsyncMutex<()>>,
91 ec2_state: Option<fakecloud_ec2::SharedEc2State>,
96}
97
98impl EfsService {
99 pub fn new(state: SharedEfsState) -> Self {
100 Self {
101 state,
102 snapshot_store: None,
103 snapshot_lock: Arc::new(AsyncMutex::new(())),
104 ec2_state: None,
105 }
106 }
107
108 pub fn with_snapshot_store(mut self, store: Arc<dyn SnapshotStore>) -> Self {
109 self.snapshot_store = Some(store);
110 self
111 }
112
113 pub fn with_ec2_state(mut self, ec2_state: fakecloud_ec2::SharedEc2State) -> Self {
114 self.ec2_state = Some(ec2_state);
115 self
116 }
117
118 pub fn snapshot_hook(&self) -> Option<fakecloud_persistence::SnapshotHook> {
124 let store = self.snapshot_store.clone()?;
125 let state = self.state.clone();
126 let lock = self.snapshot_lock.clone();
127 Some(Arc::new(move || {
128 let state = state.clone();
129 let store = store.clone();
130 let lock = lock.clone();
131 Box::pin(async move {
132 save_snapshot(&state, Some(store), &lock).await;
133 })
134 }))
135 }
136
137 fn resolve_subnet(&self, account: &str, subnet_id: &str) -> Option<(String, String, String)> {
142 let ec2 = self.ec2_state.as_ref()?;
143 let guard = ec2.read();
144 let subnet = guard.get(account)?.subnets.get(subnet_id)?;
145 Some((
146 subnet.availability_zone.clone(),
147 subnet.availability_zone_id.clone(),
148 subnet.vpc_id.clone(),
149 ))
150 }
151
152 async fn save(&self) {
153 save_snapshot(
154 &self.state,
155 self.snapshot_store.clone(),
156 &self.snapshot_lock,
157 )
158 .await;
159 }
160
161 fn resolve_action(req: &AwsRequest) -> Option<(&'static str, Option<String>)> {
164 let raw = req.raw_path.split('?').next().unwrap_or(&req.raw_path);
165 let trimmed = raw.strip_prefix('/').unwrap_or(raw);
166 let segs: Vec<String> = if trimmed.is_empty() {
172 Vec::new()
173 } else {
174 trimmed
175 .split('/')
176 .map(|s| percent_decode_str(s).decode_utf8_lossy().into_owned())
177 .collect()
178 };
179 let s: Vec<&str> = segs.iter().map(String::as_str).collect();
180 let m = &req.method;
181 let get = m == Method::GET;
182 let post = m == Method::POST;
183 let put = m == Method::PUT;
184 let del = m == Method::DELETE;
185 const V: &str = "2015-02-01";
186 let (action, label): (&'static str, Option<String>) = match s.as_slice() {
187 [v, "access-points"] if *v == V && post => ("CreateAccessPoint", None),
189 [v, "access-points"] if *v == V && get => ("DescribeAccessPoints", None),
190 [v, "access-points", id] if *v == V && del => {
191 ("DeleteAccessPoint", Some((*id).to_string()))
192 }
193 [v, "mount-targets"] if *v == V && post => ("CreateMountTarget", None),
195 [v, "mount-targets"] if *v == V && get => ("DescribeMountTargets", None),
196 [v, "mount-targets", id] if *v == V && del => {
197 ("DeleteMountTarget", Some((*id).to_string()))
198 }
199 [v, "mount-targets", id, "security-groups"] if *v == V && get => {
200 ("DescribeMountTargetSecurityGroups", Some((*id).to_string()))
201 }
202 [v, "mount-targets", id, "security-groups"] if *v == V && put => {
203 ("ModifyMountTargetSecurityGroups", Some((*id).to_string()))
204 }
205 [v, "account-preferences"] if *v == V && get => ("DescribeAccountPreferences", None),
207 [v, "account-preferences"] if *v == V && put => ("PutAccountPreferences", None),
208 [v, "file-systems"] if *v == V && post => ("CreateFileSystem", None),
210 [v, "file-systems"] if *v == V && get => ("DescribeFileSystems", None),
211 [v, "file-systems", "replication-configurations"] if *v == V && get => {
212 ("DescribeReplicationConfigurations", None)
213 }
214 [v, "file-systems", id, "replication-configuration"] if *v == V && post => {
215 ("CreateReplicationConfiguration", Some((*id).to_string()))
216 }
217 [v, "file-systems", id, "replication-configuration"] if *v == V && del => {
218 ("DeleteReplicationConfiguration", Some((*id).to_string()))
219 }
220 [v, "file-systems", id, "policy"] if *v == V && get => {
221 ("DescribeFileSystemPolicy", Some((*id).to_string()))
222 }
223 [v, "file-systems", id, "policy"] if *v == V && put => {
224 ("PutFileSystemPolicy", Some((*id).to_string()))
225 }
226 [v, "file-systems", id, "policy"] if *v == V && del => {
227 ("DeleteFileSystemPolicy", Some((*id).to_string()))
228 }
229 [v, "file-systems", id, "backup-policy"] if *v == V && get => {
230 ("DescribeBackupPolicy", Some((*id).to_string()))
231 }
232 [v, "file-systems", id, "backup-policy"] if *v == V && put => {
233 ("PutBackupPolicy", Some((*id).to_string()))
234 }
235 [v, "file-systems", id, "lifecycle-configuration"] if *v == V && get => {
236 ("DescribeLifecycleConfiguration", Some((*id).to_string()))
237 }
238 [v, "file-systems", id, "lifecycle-configuration"] if *v == V && put => {
239 ("PutLifecycleConfiguration", Some((*id).to_string()))
240 }
241 [v, "file-systems", id, "protection"] if *v == V && put => {
242 ("UpdateFileSystemProtection", Some((*id).to_string()))
243 }
244 [v, "file-systems", id] if *v == V && del => {
245 ("DeleteFileSystem", Some((*id).to_string()))
246 }
247 [v, "file-systems", id] if *v == V && put => {
248 ("UpdateFileSystem", Some((*id).to_string()))
249 }
250 [v, "create-tags", id] if *v == V && post => ("CreateTags", Some((*id).to_string())),
252 [v, "delete-tags", id] if *v == V && post => ("DeleteTags", Some((*id).to_string())),
253 [v, "tags", id] if *v == V && get => ("DescribeTags", Some((*id).to_string())),
254 [v, "resource-tags", id] if *v == V && get => {
256 ("ListTagsForResource", Some((*id).to_string()))
257 }
258 [v, "resource-tags", id] if *v == V && post => ("TagResource", Some((*id).to_string())),
259 [v, "resource-tags", id] if *v == V && del => {
260 ("UntagResource", Some((*id).to_string()))
261 }
262 _ => return None,
263 };
264 Some((action, label))
265 }
266}
267
268#[async_trait]
269impl AwsService for EfsService {
270 fn service_name(&self) -> &str {
271 "elasticfilesystem"
272 }
273
274 async fn handle(&self, req: AwsRequest) -> Result<AwsResponse, AwsServiceError> {
275 let Some((action, label)) = Self::resolve_action(&req) else {
276 return Err(AwsServiceError::aws_error(
277 StatusCode::NOT_FOUND,
278 "UnknownOperationException",
279 format!("Unknown operation: {} {}", req.method, req.raw_path),
280 ));
281 };
282 let result = self.dispatch(action, label, &req);
283 if MUTATING.contains(&action)
284 && matches!(result.as_ref(), Ok(resp) if resp.status.is_success())
285 {
286 self.save().await;
287 }
288 result
289 }
290
291 fn supported_actions(&self) -> &[&str] {
292 EFS_ACTIONS
293 }
294}
295
296struct Ctx {
298 account: String,
299 region: String,
300}
301
302impl EfsService {
303 #[allow(clippy::too_many_lines)]
304 fn dispatch(
305 &self,
306 action: &str,
307 label: Option<String>,
308 req: &AwsRequest,
309 ) -> Result<AwsResponse, AwsServiceError> {
310 let body = parse_body(req)?;
311 crate::validate::validate_input(action, &body)?;
312 let ctx = Ctx {
313 account: req.account_id.clone(),
314 region: req.region.clone(),
315 };
316 let q = parse_query(&req.raw_query);
317 crate::validate::validate_query(&q)?;
318 let label = label.unwrap_or_default();
319 match action {
320 "CreateFileSystem" => self.create_file_system(&ctx, &body),
321 "DescribeFileSystems" => self.describe_file_systems(&ctx, &q),
322 "DeleteFileSystem" => self.delete_file_system(&ctx, &label),
323 "UpdateFileSystem" => self.update_file_system(&ctx, &label, &body),
324 "UpdateFileSystemProtection" => self.update_file_system_protection(&ctx, &label, &body),
325 "CreateMountTarget" => self.create_mount_target(&ctx, &body),
326 "DescribeMountTargets" => self.describe_mount_targets(&ctx, &q),
327 "DeleteMountTarget" => self.delete_mount_target(&ctx, &label),
328 "DescribeMountTargetSecurityGroups" => self.describe_mt_security_groups(&ctx, &label),
329 "ModifyMountTargetSecurityGroups" => {
330 self.modify_mt_security_groups(&ctx, &label, &body)
331 }
332 "CreateAccessPoint" => self.create_access_point(&ctx, &body),
333 "DescribeAccessPoints" => self.describe_access_points(&ctx, &q),
334 "DeleteAccessPoint" => self.delete_access_point(&ctx, &label),
335 "PutLifecycleConfiguration" => self.put_lifecycle_configuration(&ctx, &label, &body),
336 "DescribeLifecycleConfiguration" => self.describe_lifecycle_configuration(&ctx, &label),
337 "PutBackupPolicy" => self.put_backup_policy(&ctx, &label, &body),
338 "DescribeBackupPolicy" => self.describe_backup_policy(&ctx, &label),
339 "PutFileSystemPolicy" => self.put_file_system_policy(&ctx, &label, &body),
340 "DescribeFileSystemPolicy" => self.describe_file_system_policy(&ctx, &label),
341 "DeleteFileSystemPolicy" => self.delete_file_system_policy(&ctx, &label),
342 "CreateReplicationConfiguration" => {
343 self.create_replication_configuration(&ctx, &label, &body)
344 }
345 "DescribeReplicationConfigurations" => {
346 self.describe_replication_configurations(&ctx, &q)
347 }
348 "DeleteReplicationConfiguration" => self.delete_replication_configuration(&ctx, &label),
349 "CreateTags" => self.create_tags(&ctx, &label, &body),
350 "DeleteTags" => self.delete_tags(&ctx, &label, &body),
351 "DescribeTags" => self.describe_tags(&ctx, &label, &q),
352 "TagResource" => self.tag_resource(&ctx, &label, &body),
353 "UntagResource" => self.untag_resource(&ctx, &label, &q),
354 "ListTagsForResource" => self.list_tags_for_resource(&ctx, &label, &q),
355 "DescribeAccountPreferences" => self.describe_account_preferences(&ctx),
356 "PutAccountPreferences" => self.put_account_preferences(&ctx, &body),
357 _ => Err(AwsServiceError::action_not_implemented(
358 "elasticfilesystem",
359 action,
360 )),
361 }
362 }
363}
364
365fn ok(status: StatusCode, v: Value) -> Result<AwsResponse, AwsServiceError> {
368 Ok(AwsResponse::json_value(status, v))
369}
370
371fn empty(status: StatusCode) -> Result<AwsResponse, AwsServiceError> {
372 Ok(AwsResponse::json_value(status, json!({})))
373}
374
375fn parse_body(req: &AwsRequest) -> Result<Value, AwsServiceError> {
376 if req.body.is_empty() {
377 return Ok(json!({}));
378 }
379 serde_json::from_slice(&req.body)
380 .map_err(|e| bad_request(&format!("Request body is malformed: {e}")))
381}
382
383fn bad_request(msg: &str) -> AwsServiceError {
384 AwsServiceError::aws_error(StatusCode::BAD_REQUEST, "BadRequest", msg)
385}
386
387fn fs_not_found(id: &str) -> AwsServiceError {
388 AwsServiceError::aws_error(
389 StatusCode::NOT_FOUND,
390 "FileSystemNotFound",
391 format!("File system '{id}' does not exist."),
392 )
393}
394
395fn mt_not_found(id: &str) -> AwsServiceError {
396 AwsServiceError::aws_error(
397 StatusCode::NOT_FOUND,
398 "MountTargetNotFound",
399 format!("Mount target '{id}' does not exist."),
400 )
401}
402
403fn ap_not_found(id: &str) -> AwsServiceError {
404 AwsServiceError::aws_error(
405 StatusCode::NOT_FOUND,
406 "AccessPointNotFound",
407 format!("Access point '{id}' does not exist."),
408 )
409}
410
411fn subnet_not_found(id: &str) -> AwsServiceError {
412 AwsServiceError::aws_error(
413 StatusCode::BAD_REQUEST,
414 "SubnetNotFound",
415 format!("The subnet ID '{id}' is invalid or does not exist."),
416 )
417}
418
419fn replication_not_found(id: &str) -> AwsServiceError {
420 AwsServiceError::aws_error(
421 StatusCode::NOT_FOUND,
422 "ReplicationNotFound",
423 format!("Replication configuration for file system '{id}' does not exist."),
424 )
425}
426
427fn incorrect_fs_lifecycle_state(id: &str, state: &str) -> AwsServiceError {
428 AwsServiceError::aws_error(
429 StatusCode::CONFLICT,
430 "IncorrectFileSystemLifeCycleState",
431 format!(
432 "File system '{id}' is in life cycle state '{state}' but must be 'available' for this operation."
433 ),
434 )
435}
436
437fn conflict(code: &str, msg: &str) -> AwsServiceError {
438 AwsServiceError::aws_error(StatusCode::CONFLICT, code, msg)
439}
440
441fn hex17() -> String {
443 Uuid::new_v4()
444 .simple()
445 .to_string()
446 .chars()
447 .filter(char::is_ascii_hexdigit)
448 .take(17)
449 .collect()
450}
451
452fn now_ts() -> f64 {
453 chrono::Utc::now().timestamp() as f64
454}
455
456fn fs_arn(ctx: &Ctx, fsid: &str) -> String {
457 format!(
458 "arn:aws:elasticfilesystem:{}:{}:file-system/{}",
459 ctx.region, ctx.account, fsid
460 )
461}
462
463fn ap_arn(ctx: &Ctx, apid: &str) -> String {
464 format!(
465 "arn:aws:elasticfilesystem:{}:{}:access-point/{}",
466 ctx.region, ctx.account, apid
467 )
468}
469
470fn normalize_fs_id(raw: &str) -> String {
473 raw.rsplit('/').next().unwrap_or(raw).to_string()
474}
475
476fn hash_str(s: &str) -> u64 {
478 let mut h: u64 = 0xcbf2_9ce4_8422_2325;
479 for b in s.as_bytes() {
480 h ^= *b as u64;
481 h = h.wrapping_mul(0x0000_0100_0000_01b3);
482 }
483 h
484}
485
486fn parse_query(raw: &str) -> Vec<(String, String)> {
487 raw.split('&')
488 .filter(|p| !p.is_empty())
489 .map(|pair| {
490 let (k, v) = pair.split_once('=').unwrap_or((pair, ""));
491 (
492 percent_decode_str(k).decode_utf8_lossy().into_owned(),
493 percent_decode_str(v).decode_utf8_lossy().into_owned(),
494 )
495 })
496 .collect()
497}
498
499fn query_one<'a>(q: &'a [(String, String)], key: &str) -> Option<&'a str> {
500 q.iter()
501 .find(|(k, _)| k == key)
502 .map(|(_, v)| v.as_str())
503 .filter(|v| !v.is_empty())
504}
505
506fn query_all(q: &[(String, String)], key: &str) -> Vec<String> {
507 q.iter()
508 .filter(|(k, _)| k == key)
509 .map(|(_, v)| v.clone())
510 .collect()
511}
512
513fn tags_to_map(b: &Value) -> Map<String, Value> {
515 let mut out = Map::new();
516 if let Some(arr) = b.get("Tags").and_then(Value::as_array) {
517 for t in arr {
518 if let (Some(k), Some(v)) = (
519 t.get("Key").and_then(Value::as_str),
520 t.get("Value").and_then(Value::as_str),
521 ) {
522 out.insert(k.to_string(), json!(v));
523 }
524 }
525 }
526 out
527}
528
529fn tags_list(data: &EfsData, resource_id: &str) -> Value {
531 let arr: Vec<Value> = data
532 .tags
533 .get(resource_id)
534 .map(|m| {
535 m.iter()
536 .map(|(k, v)| json!({ "Key": k, "Value": v }))
537 .collect()
538 })
539 .unwrap_or_default();
540 Value::Array(arr)
541}
542
543impl EfsService {
544 fn account<'g>(
545 &self,
546 guard: &'g mut RwLockWriteGuard<'_, MultiAccountState<EfsData>>,
547 ctx: &Ctx,
548 ) -> &'g mut EfsData {
549 guard.get_or_create(&ctx.account)
550 }
551}
552
553impl EfsService {
556 fn create_file_system(&self, ctx: &Ctx, b: &Value) -> Result<AwsResponse, AwsServiceError> {
557 let creation_token = b
558 .get("CreationToken")
559 .and_then(Value::as_str)
560 .unwrap_or_default()
561 .to_string();
562 let mut guard = self.state.write();
563 let data = guard.get_or_create(&ctx.account);
564 if let Some((existing_id, _)) = data.file_systems.iter().find(|(_, fs)| {
567 fs.get("CreationToken").and_then(Value::as_str) == Some(creation_token.as_str())
568 }) {
569 let existing_id = existing_id.clone();
570 return Err(AwsServiceError::aws_error_with_fields(
571 StatusCode::CONFLICT,
572 "FileSystemAlreadyExists",
573 format!("File system already exists with creation token {creation_token}"),
574 vec![("FileSystemId".to_string(), existing_id)],
575 ));
576 }
577
578 let fsid = format!("fs-{}", hex17());
579 let performance_mode = b
580 .get("PerformanceMode")
581 .and_then(Value::as_str)
582 .unwrap_or("generalPurpose");
583 let throughput_mode = b
584 .get("ThroughputMode")
585 .and_then(Value::as_str)
586 .unwrap_or("bursting");
587 if throughput_mode == "provisioned" {
591 match b
592 .get("ProvisionedThroughputInMibps")
593 .and_then(Value::as_f64)
594 {
595 None => {
596 return Err(bad_request(
597 "ProvisionedThroughputInMibps is required when ThroughputMode is set to provisioned.",
598 ));
599 }
600 Some(v) if v < 1.0 => {
601 return Err(bad_request(
602 "Value at 'ProvisionedThroughputInMibps' failed to satisfy constraint: Member must have value greater than or equal to 1",
603 ));
604 }
605 Some(_) => {}
606 }
607 } else if b.get("ProvisionedThroughputInMibps").is_some() {
608 return Err(bad_request(
609 "ProvisionedThroughputInMibps is only applicable when ThroughputMode is set to provisioned.",
610 ));
611 }
612 let encrypted = b.get("Encrypted").and_then(Value::as_bool).unwrap_or(false);
613
614 let mut fs = Map::new();
615 fs.insert("OwnerId".into(), json!(ctx.account));
616 fs.insert("CreationToken".into(), json!(creation_token));
617 fs.insert("FileSystemId".into(), json!(fsid));
618 fs.insert("FileSystemArn".into(), json!(fs_arn(ctx, &fsid)));
619 fs.insert("CreationTime".into(), json!(now_ts()));
620 fs.insert("LifeCycleState".into(), json!("creating"));
623 fs.insert("NumberOfMountTargets".into(), json!(0));
624 fs.insert(
625 "SizeInBytes".into(),
626 json!({
627 "Value": 6144,
628 "Timestamp": now_ts(),
629 "ValueInIA": 0,
630 "ValueInStandard": 6144,
631 "ValueInArchive": 0
632 }),
633 );
634 fs.insert("PerformanceMode".into(), json!(performance_mode));
635 fs.insert("ThroughputMode".into(), json!(throughput_mode));
636 fs.insert("Encrypted".into(), json!(encrypted));
637 if encrypted {
638 let kms = b
639 .get("KmsKeyId")
640 .and_then(Value::as_str)
641 .map(str::to_string)
642 .unwrap_or_else(|| {
643 format!(
644 "arn:aws:kms:{}:{}:key/{}-{}-{}-{}-{}",
645 ctx.region,
646 ctx.account,
647 &hex17()[..8],
648 &hex17()[..4],
649 &hex17()[..4],
650 &hex17()[..4],
651 &hex17()[..12.min(hex17().len())]
652 )
653 });
654 fs.insert("KmsKeyId".into(), json!(kms));
655 }
656 if throughput_mode == "provisioned" {
657 if let Some(p) = b.get("ProvisionedThroughputInMibps") {
658 fs.insert("ProvisionedThroughputInMibps".into(), p.clone());
659 }
660 }
661 if let Some(az) = b.get("AvailabilityZoneName").and_then(Value::as_str) {
662 fs.insert("AvailabilityZoneName".into(), json!(az));
663 fs.insert(
664 "AvailabilityZoneId".into(),
665 json!(format!("{}-az1", ctx.region)),
666 );
667 }
668 let tag_map = tags_to_map(b);
670 if let Some(name) = tag_map.get("Name").and_then(Value::as_str) {
671 fs.insert("Name".into(), json!(name));
672 }
673 fs.insert("Tags".into(), tags_list_from_map(&tag_map));
674 fs.insert(
675 "FileSystemProtection".into(),
676 json!({ "ReplicationOverwriteProtection": "ENABLED" }),
677 );
678
679 if !tag_map.is_empty() {
680 let entry = data.tags.entry(fsid.clone()).or_default();
681 for (k, v) in &tag_map {
682 if let Some(vs) = v.as_str() {
683 entry.insert(k.clone(), vs.to_string());
684 }
685 }
686 }
687 data.file_systems
688 .insert(fsid.clone(), Value::Object(fs.clone()));
689 ok(StatusCode::CREATED, Value::Object(fs))
690 }
691
692 fn describe_file_systems(
693 &self,
694 ctx: &Ctx,
695 q: &[(String, String)],
696 ) -> Result<AwsResponse, AwsServiceError> {
697 let filter_id = query_one(q, "FileSystemId").map(normalize_fs_id);
698 let filter_token = query_one(q, "CreationToken");
699 let mut guard = self.state.write();
700 let data = self.account(&mut guard, ctx);
701 data.reconcile_lifecycle();
702
703 if let Some(id) = &filter_id {
704 if !data.file_systems.contains_key(id) {
705 return Err(fs_not_found(id));
706 }
707 }
708 let rows: Vec<Value> = data
709 .file_systems
710 .values()
711 .filter(|fs| {
712 filter_id
713 .as_ref()
714 .map(|id| fs.get("FileSystemId").and_then(Value::as_str) == Some(id.as_str()))
715 .unwrap_or(true)
716 })
717 .filter(|fs| {
718 filter_token
719 .map(|t| fs.get("CreationToken").and_then(Value::as_str) == Some(t))
720 .unwrap_or(true)
721 })
722 .map(|fs| with_live_mount_count(fs, data))
723 .collect();
724
725 let (page, next) =
727 paginate_marker(rows, query_one(q, "Marker"), query_one(q, "MaxItems"), 100);
728 let mut out = Map::new();
729 out.insert("FileSystems".into(), Value::Array(page));
730 if let Some(n) = next {
731 out.insert("NextMarker".into(), json!(n));
732 }
733 ok(StatusCode::OK, Value::Object(out))
734 }
735
736 fn require_fs<'a>(
737 &self,
738 data: &'a mut EfsData,
739 fsid: &str,
740 ) -> Result<&'a mut Value, AwsServiceError> {
741 data.file_systems
742 .get_mut(fsid)
743 .ok_or_else(|| fs_not_found(fsid))
744 }
745
746 fn delete_file_system(&self, ctx: &Ctx, label: &str) -> Result<AwsResponse, AwsServiceError> {
747 let fsid = normalize_fs_id(label);
748 let mut guard = self.state.write();
749 let data = self.account(&mut guard, ctx);
750 if !data.file_systems.contains_key(&fsid) {
751 return Err(fs_not_found(&fsid));
752 }
753 let has_mt = data
757 .mount_targets
758 .values()
759 .any(|mt| mt.get("FileSystemId").and_then(Value::as_str) == Some(fsid.as_str()));
760 let has_ap = data.access_points.values().any(|ap| {
761 normalize_fs_id(ap.get("FileSystemId").and_then(Value::as_str).unwrap_or("")) == fsid
762 });
763 if has_mt || has_ap {
764 return Err(conflict(
765 "FileSystemInUse",
766 &format!("File system '{fsid}' is in use and cannot be deleted."),
767 ));
768 }
769 data.file_systems.remove(&fsid);
770 data.tags.remove(&fsid);
771 data.lifecycle_configs.remove(&fsid);
772 data.backup_policies.remove(&fsid);
773 data.file_system_policies.remove(&fsid);
774 data.replications.remove(&fsid);
775 empty(StatusCode::NO_CONTENT)
776 }
777
778 fn update_file_system(
779 &self,
780 ctx: &Ctx,
781 label: &str,
782 b: &Value,
783 ) -> Result<AwsResponse, AwsServiceError> {
784 let fsid = normalize_fs_id(label);
785 let mut guard = self.state.write();
786 let data = self.account(&mut guard, ctx);
787 let fs = self.require_fs(data, &fsid)?;
788 let obj = fs.as_object_mut().unwrap();
789 if obj.get("LifeCycleState").and_then(Value::as_str) == Some("creating") {
790 obj.insert("LifeCycleState".into(), json!("available"));
791 }
792 if let Some(tm) = b.get("ThroughputMode") {
793 obj.insert("ThroughputMode".into(), tm.clone());
794 }
795 if let Some(p) = b.get("ProvisionedThroughputInMibps") {
796 obj.insert("ProvisionedThroughputInMibps".into(), p.clone());
797 }
798 if obj.get("ThroughputMode").and_then(Value::as_str) != Some("provisioned") {
800 obj.remove("ProvisionedThroughputInMibps");
801 }
802 let fs_snapshot = fs.clone();
803 let updated = with_live_mount_count(&fs_snapshot, data);
804 ok(StatusCode::ACCEPTED, updated)
805 }
806
807 fn update_file_system_protection(
808 &self,
809 ctx: &Ctx,
810 label: &str,
811 b: &Value,
812 ) -> Result<AwsResponse, AwsServiceError> {
813 let fsid = normalize_fs_id(label);
814 let mut guard = self.state.write();
815 let data = self.account(&mut guard, ctx);
816 let fs = self.require_fs(data, &fsid)?;
817 let protection = b
818 .get("ReplicationOverwriteProtection")
819 .and_then(Value::as_str)
820 .unwrap_or("ENABLED");
821 fs.as_object_mut().unwrap().insert(
822 "FileSystemProtection".into(),
823 json!({ "ReplicationOverwriteProtection": protection }),
824 );
825 ok(
826 StatusCode::OK,
827 json!({ "ReplicationOverwriteProtection": protection }),
828 )
829 }
830}
831
832fn tags_list_from_map(m: &Map<String, Value>) -> Value {
833 Value::Array(
834 m.iter()
835 .map(|(k, v)| json!({ "Key": k, "Value": v }))
836 .collect(),
837 )
838}
839
840fn with_live_mount_count(fs: &Value, data: &EfsData) -> Value {
843 let mut obj = fs.as_object().cloned().unwrap_or_default();
844 let fsid = obj
845 .get("FileSystemId")
846 .and_then(Value::as_str)
847 .unwrap_or("");
848 let count = data
849 .mount_targets
850 .values()
851 .filter(|mt| mt.get("FileSystemId").and_then(Value::as_str) == Some(fsid))
852 .count();
853 obj.insert("NumberOfMountTargets".into(), json!(count));
854 Value::Object(obj)
855}
856
857fn paginate_marker(
862 rows: Vec<Value>,
863 marker: Option<&str>,
864 max: Option<&str>,
865 default_max: usize,
866) -> (Vec<Value>, Option<String>) {
867 let start = marker.and_then(|m| m.parse::<usize>().ok()).unwrap_or(0);
868 let max = max
869 .and_then(|m| m.parse::<usize>().ok())
870 .map(|m| m.max(1))
871 .unwrap_or(default_max);
872 let end = start.saturating_add(max).min(rows.len());
873 let page = rows.get(start..end).unwrap_or(&[]).to_vec();
874 let next = if end < rows.len() {
875 Some(end.to_string())
876 } else {
877 None
878 };
879 (page, next)
880}
881
882impl EfsService {
885 fn create_mount_target(&self, ctx: &Ctx, b: &Value) -> Result<AwsResponse, AwsServiceError> {
886 let fsid = normalize_fs_id(b.get("FileSystemId").and_then(Value::as_str).unwrap_or(""));
887 let subnet_id = b
888 .get("SubnetId")
889 .and_then(Value::as_str)
890 .unwrap_or("")
891 .to_string();
892 let h = hash_str(&subnet_id);
897 let (az_name, az_id, vpc_id) = match self.resolve_subnet(&ctx.account, &subnet_id) {
898 Some((az, azid, vpc)) => (az, azid, vpc),
899 None => {
900 if self.ec2_state.is_some() {
905 return Err(subnet_not_found(&subnet_id));
906 }
907 let az_index = (h % 3) as u8;
908 (
909 format!("{}{}", ctx.region, (b'a' + az_index) as char),
910 format!("{}-az{}", ctx.region, az_index + 1),
911 format!("vpc-{:017x}", h & 0x000f_ffff_ffff_ffff),
912 )
913 }
914 };
915 let mut guard = self.state.write();
916 let data = self.account(&mut guard, ctx);
917 let fs = data
923 .file_systems
924 .get(&fsid)
925 .ok_or_else(|| fs_not_found(&fsid))?;
926 let fs_state = fs
927 .get("LifeCycleState")
928 .and_then(Value::as_str)
929 .unwrap_or("");
930 if fs_state != "available" {
931 return Err(incorrect_fs_lifecycle_state(&fsid, fs_state));
932 }
933 for mt in data.mount_targets.values() {
935 if mt.get("FileSystemId").and_then(Value::as_str) == Some(fsid.as_str())
936 && mt.get("AvailabilityZoneName").and_then(Value::as_str) == Some(az_name.as_str())
937 {
938 return Err(conflict(
939 "MountTargetConflict",
940 "A mount target already exists in this Availability Zone for the file system.",
941 ));
942 }
943 }
944
945 let mtid = format!("fsmt-{}", hex17());
946 let ip = b
947 .get("IpAddress")
948 .and_then(Value::as_str)
949 .map(str::to_string)
950 .unwrap_or_else(|| format!("10.0.{}.{}", (h >> 8) % 256, h % 254 + 1));
951 let eni_id = format!("eni-{}", hex17());
952 let security_groups: Vec<String> = b
953 .get("SecurityGroups")
954 .and_then(Value::as_array)
955 .map(|a| {
956 a.iter()
957 .filter_map(|v| v.as_str().map(str::to_string))
958 .collect()
959 })
960 .filter(|v: &Vec<String>| !v.is_empty())
961 .unwrap_or_else(|| vec![format!("sg-{:017x}", h & 0x000f_ffff_ffff_ffff)]);
962
963 let mut mt = Map::new();
964 mt.insert("OwnerId".into(), json!(ctx.account));
965 mt.insert("MountTargetId".into(), json!(mtid));
966 mt.insert("FileSystemId".into(), json!(fsid));
967 mt.insert("SubnetId".into(), json!(subnet_id));
968 mt.insert("LifeCycleState".into(), json!("creating"));
969 mt.insert("IpAddress".into(), json!(ip));
970 mt.insert("NetworkInterfaceId".into(), json!(eni_id));
971 mt.insert("AvailabilityZoneId".into(), json!(az_id));
972 mt.insert("AvailabilityZoneName".into(), json!(az_name));
973 mt.insert("VpcId".into(), json!(vpc_id));
974 if let Some(ip6) = b.get("Ipv6Address") {
975 mt.insert("Ipv6Address".into(), ip6.clone());
976 }
977
978 data.mount_target_security_groups
979 .insert(mtid.clone(), security_groups);
980 data.mount_targets
981 .insert(mtid.clone(), Value::Object(mt.clone()));
982 ok(StatusCode::OK, Value::Object(mt))
983 }
984
985 fn describe_mount_targets(
986 &self,
987 ctx: &Ctx,
988 q: &[(String, String)],
989 ) -> Result<AwsResponse, AwsServiceError> {
990 let by_fs = query_one(q, "FileSystemId").map(normalize_fs_id);
991 let by_mt = query_one(q, "MountTargetId");
992 let by_ap = query_one(q, "AccessPointId");
993 let filter_count = usize::from(by_fs.is_some())
996 + usize::from(by_mt.is_some())
997 + usize::from(by_ap.is_some());
998 if filter_count != 1 {
999 return Err(bad_request(
1000 "Exactly one of FileSystemId, MountTargetId, or AccessPointId must be specified.",
1001 ));
1002 }
1003 let mut guard = self.state.write();
1004 let data = self.account(&mut guard, ctx);
1005 data.reconcile_lifecycle();
1006
1007 let ap_fs = if let Some(ap) = by_ap {
1009 match data.access_points.get(ap) {
1010 Some(ap_obj) => Some(normalize_fs_id(
1011 ap_obj
1012 .get("FileSystemId")
1013 .and_then(Value::as_str)
1014 .unwrap_or(""),
1015 )),
1016 None => return Err(ap_not_found(ap)),
1017 }
1018 } else {
1019 None
1020 };
1021 if let Some(fsid) = &by_fs {
1022 if !data.file_systems.contains_key(fsid) {
1023 return Err(fs_not_found(fsid));
1024 }
1025 }
1026 if let Some(mtid) = by_mt {
1027 if !data.mount_targets.contains_key(mtid) {
1028 return Err(mt_not_found(mtid));
1029 }
1030 }
1031
1032 let rows: Vec<Value> = data
1033 .mount_targets
1034 .iter()
1035 .filter(|(mtid, mt)| {
1036 let fs = mt.get("FileSystemId").and_then(Value::as_str);
1037 by_fs.as_deref().map(|f| fs == Some(f)).unwrap_or(true)
1038 && by_mt.map(|m| mtid.as_str() == m).unwrap_or(true)
1039 && ap_fs.as_deref().map(|f| fs == Some(f)).unwrap_or(true)
1040 })
1041 .map(|(_, mt)| mt.clone())
1042 .collect();
1043
1044 let (page, next) =
1046 paginate_marker(rows, query_one(q, "Marker"), query_one(q, "MaxItems"), 10);
1047 let mut out = Map::new();
1048 out.insert("MountTargets".into(), Value::Array(page));
1049 if let Some(n) = next {
1050 out.insert("NextMarker".into(), json!(n));
1051 }
1052 ok(StatusCode::OK, Value::Object(out))
1053 }
1054
1055 fn delete_mount_target(&self, ctx: &Ctx, label: &str) -> Result<AwsResponse, AwsServiceError> {
1056 let mut guard = self.state.write();
1057 let data = self.account(&mut guard, ctx);
1058 if data.mount_targets.remove(label).is_none() {
1059 return Err(mt_not_found(label));
1060 }
1061 data.mount_target_security_groups.remove(label);
1062 empty(StatusCode::NO_CONTENT)
1063 }
1064
1065 fn describe_mt_security_groups(
1066 &self,
1067 ctx: &Ctx,
1068 label: &str,
1069 ) -> Result<AwsResponse, AwsServiceError> {
1070 let mut guard = self.state.write();
1071 let data = self.account(&mut guard, ctx);
1072 data.reconcile_lifecycle();
1073 if !data.mount_targets.contains_key(label) {
1074 return Err(mt_not_found(label));
1075 }
1076 let sgs = data
1077 .mount_target_security_groups
1078 .get(label)
1079 .cloned()
1080 .unwrap_or_default();
1081 ok(StatusCode::OK, json!({ "SecurityGroups": sgs }))
1082 }
1083
1084 fn modify_mt_security_groups(
1085 &self,
1086 ctx: &Ctx,
1087 label: &str,
1088 b: &Value,
1089 ) -> Result<AwsResponse, AwsServiceError> {
1090 let mut guard = self.state.write();
1091 let data = self.account(&mut guard, ctx);
1092 if !data.mount_targets.contains_key(label) {
1093 return Err(mt_not_found(label));
1094 }
1095 let sgs: Vec<String> = b
1096 .get("SecurityGroups")
1097 .and_then(Value::as_array)
1098 .map(|a| {
1099 a.iter()
1100 .filter_map(|v| v.as_str().map(str::to_string))
1101 .collect()
1102 })
1103 .unwrap_or_default();
1104 data.mount_target_security_groups
1105 .insert(label.to_string(), sgs);
1106 empty(StatusCode::NO_CONTENT)
1107 }
1108}
1109
1110impl EfsService {
1113 fn create_access_point(&self, ctx: &Ctx, b: &Value) -> Result<AwsResponse, AwsServiceError> {
1114 let client_token = b
1115 .get("ClientToken")
1116 .and_then(Value::as_str)
1117 .unwrap_or_default()
1118 .to_string();
1119 let fsid = normalize_fs_id(b.get("FileSystemId").and_then(Value::as_str).unwrap_or(""));
1120 let mut guard = self.state.write();
1121 let data = self.account(&mut guard, ctx);
1122 if !data.file_systems.contains_key(&fsid) {
1123 return Err(fs_not_found(&fsid));
1124 }
1125 if let Some((existing_id, _)) = data.access_points.iter().find(|(_, ap)| {
1126 ap.get("ClientToken").and_then(Value::as_str) == Some(client_token.as_str())
1127 }) {
1128 let existing_id = existing_id.clone();
1129 return Err(AwsServiceError::aws_error_with_fields(
1130 StatusCode::CONFLICT,
1131 "AccessPointAlreadyExists",
1132 format!("Access point already exists with client token {client_token}"),
1133 vec![("AccessPointId".to_string(), existing_id)],
1134 ));
1135 }
1136
1137 let apid = format!("fsap-{}", hex17());
1138 let tag_map = tags_to_map(b);
1139 let mut ap = Map::new();
1140 ap.insert("ClientToken".into(), json!(client_token));
1141 if let Some(name) = tag_map.get("Name").and_then(Value::as_str) {
1142 ap.insert("Name".into(), json!(name));
1143 }
1144 ap.insert("Tags".into(), tags_list_from_map(&tag_map));
1145 ap.insert("AccessPointId".into(), json!(apid));
1146 ap.insert("AccessPointArn".into(), json!(ap_arn(ctx, &apid)));
1147 ap.insert("FileSystemId".into(), json!(fsid));
1148 if let Some(pu) = b.get("PosixUser") {
1149 ap.insert("PosixUser".into(), pu.clone());
1150 }
1151 if let Some(rd) = b.get("RootDirectory") {
1152 ap.insert("RootDirectory".into(), rd.clone());
1153 } else {
1154 ap.insert("RootDirectory".into(), json!({ "Path": "/" }));
1155 }
1156 ap.insert("OwnerId".into(), json!(ctx.account));
1157 ap.insert("LifeCycleState".into(), json!("creating"));
1158
1159 if !tag_map.is_empty() {
1160 let entry = data.tags.entry(apid.clone()).or_default();
1161 for (k, v) in &tag_map {
1162 if let Some(vs) = v.as_str() {
1163 entry.insert(k.clone(), vs.to_string());
1164 }
1165 }
1166 }
1167 data.access_points
1168 .insert(apid.clone(), Value::Object(ap.clone()));
1169 ok(StatusCode::OK, Value::Object(ap))
1170 }
1171
1172 fn describe_access_points(
1173 &self,
1174 ctx: &Ctx,
1175 q: &[(String, String)],
1176 ) -> Result<AwsResponse, AwsServiceError> {
1177 let by_ap = query_one(q, "AccessPointId");
1178 let by_fs = query_one(q, "FileSystemId").map(normalize_fs_id);
1179 let mut guard = self.state.write();
1180 let data = self.account(&mut guard, ctx);
1181 data.reconcile_lifecycle();
1182
1183 if let Some(ap) = by_ap {
1184 if !data.access_points.contains_key(ap) {
1185 return Err(ap_not_found(ap));
1186 }
1187 }
1188 if let Some(fsid) = &by_fs {
1189 if !data.file_systems.contains_key(fsid) {
1190 return Err(fs_not_found(fsid));
1191 }
1192 }
1193 let rows: Vec<Value> = data
1194 .access_points
1195 .iter()
1196 .filter(|(apid, ap)| {
1197 by_ap.map(|a| apid.as_str() == a).unwrap_or(true)
1198 && by_fs
1199 .as_deref()
1200 .map(|f| ap.get("FileSystemId").and_then(Value::as_str) == Some(f))
1201 .unwrap_or(true)
1202 })
1203 .map(|(_, ap)| ap.clone())
1204 .collect();
1205
1206 let start = query_one(q, "NextToken")
1208 .and_then(|t| t.parse::<usize>().ok())
1209 .unwrap_or(0);
1210 let max = query_one(q, "MaxResults")
1211 .and_then(|m| m.parse::<usize>().ok())
1212 .map(|m| m.max(1))
1213 .unwrap_or(100);
1214 let end = start.saturating_add(max).min(rows.len());
1215 let page = rows.get(start..end).unwrap_or(&[]).to_vec();
1216 let mut out = Map::new();
1217 out.insert("AccessPoints".into(), Value::Array(page));
1218 if end < rows.len() {
1219 out.insert("NextToken".into(), json!(end.to_string()));
1220 }
1221 ok(StatusCode::OK, Value::Object(out))
1222 }
1223
1224 fn delete_access_point(&self, ctx: &Ctx, label: &str) -> Result<AwsResponse, AwsServiceError> {
1225 let mut guard = self.state.write();
1226 let data = self.account(&mut guard, ctx);
1227 if data.access_points.remove(label).is_none() {
1228 return Err(ap_not_found(label));
1229 }
1230 data.tags.remove(label);
1231 empty(StatusCode::NO_CONTENT)
1232 }
1233}
1234
1235impl EfsService {
1238 fn put_lifecycle_configuration(
1239 &self,
1240 ctx: &Ctx,
1241 label: &str,
1242 b: &Value,
1243 ) -> Result<AwsResponse, AwsServiceError> {
1244 let fsid = normalize_fs_id(label);
1245 let mut guard = self.state.write();
1246 let data = self.account(&mut guard, ctx);
1247 if !data.file_systems.contains_key(&fsid) {
1248 return Err(fs_not_found(&fsid));
1249 }
1250 let policies = b.get("LifecyclePolicies").cloned().unwrap_or(json!([]));
1251 data.lifecycle_configs.insert(fsid, policies.clone());
1252 ok(StatusCode::OK, json!({ "LifecyclePolicies": policies }))
1253 }
1254
1255 fn describe_lifecycle_configuration(
1256 &self,
1257 ctx: &Ctx,
1258 label: &str,
1259 ) -> Result<AwsResponse, AwsServiceError> {
1260 let fsid = normalize_fs_id(label);
1261 let guard = self.state.read();
1262 let data = guard.get(&ctx.account);
1263 let exists = data
1264 .map(|d| d.file_systems.contains_key(&fsid))
1265 .unwrap_or(false);
1266 if !exists {
1267 return Err(fs_not_found(&fsid));
1268 }
1269 let policies = data
1270 .and_then(|d| d.lifecycle_configs.get(&fsid).cloned())
1271 .unwrap_or(json!([]));
1272 ok(StatusCode::OK, json!({ "LifecyclePolicies": policies }))
1273 }
1274
1275 fn put_backup_policy(
1276 &self,
1277 ctx: &Ctx,
1278 label: &str,
1279 b: &Value,
1280 ) -> Result<AwsResponse, AwsServiceError> {
1281 let fsid = normalize_fs_id(label);
1282 let mut guard = self.state.write();
1283 let data = self.account(&mut guard, ctx);
1284 if !data.file_systems.contains_key(&fsid) {
1285 return Err(fs_not_found(&fsid));
1286 }
1287 let status = b
1288 .get("BackupPolicy")
1289 .and_then(|p| p.get("Status"))
1290 .and_then(Value::as_str)
1291 .unwrap_or("DISABLED")
1292 .to_string();
1293 data.backup_policies.insert(fsid, status.clone());
1294 ok(
1295 StatusCode::OK,
1296 json!({ "BackupPolicy": { "Status": status } }),
1297 )
1298 }
1299
1300 fn describe_backup_policy(
1301 &self,
1302 ctx: &Ctx,
1303 label: &str,
1304 ) -> Result<AwsResponse, AwsServiceError> {
1305 let fsid = normalize_fs_id(label);
1306 let guard = self.state.read();
1307 let data = guard.get(&ctx.account);
1308 let exists = data
1309 .map(|d| d.file_systems.contains_key(&fsid))
1310 .unwrap_or(false);
1311 if !exists {
1312 return Err(fs_not_found(&fsid));
1313 }
1314 let status = data
1315 .and_then(|d| d.backup_policies.get(&fsid).cloned())
1316 .unwrap_or_else(|| "DISABLED".to_string());
1317 ok(
1318 StatusCode::OK,
1319 json!({ "BackupPolicy": { "Status": status } }),
1320 )
1321 }
1322
1323 fn put_file_system_policy(
1324 &self,
1325 ctx: &Ctx,
1326 label: &str,
1327 b: &Value,
1328 ) -> Result<AwsResponse, AwsServiceError> {
1329 let fsid = normalize_fs_id(label);
1330 let mut guard = self.state.write();
1331 let data = self.account(&mut guard, ctx);
1332 if !data.file_systems.contains_key(&fsid) {
1333 return Err(fs_not_found(&fsid));
1334 }
1335 let policy = b
1336 .get("Policy")
1337 .and_then(Value::as_str)
1338 .unwrap_or_default()
1339 .to_string();
1340 data.file_system_policies
1341 .insert(fsid.clone(), policy.clone());
1342 ok(
1343 StatusCode::OK,
1344 json!({ "FileSystemId": fsid, "Policy": policy }),
1345 )
1346 }
1347
1348 fn describe_file_system_policy(
1349 &self,
1350 ctx: &Ctx,
1351 label: &str,
1352 ) -> Result<AwsResponse, AwsServiceError> {
1353 let fsid = normalize_fs_id(label);
1354 let guard = self.state.read();
1355 let data = guard.get(&ctx.account);
1356 let exists = data
1357 .map(|d| d.file_systems.contains_key(&fsid))
1358 .unwrap_or(false);
1359 if !exists {
1360 return Err(fs_not_found(&fsid));
1361 }
1362 match data.and_then(|d| d.file_system_policies.get(&fsid).cloned()) {
1363 Some(policy) => ok(
1364 StatusCode::OK,
1365 json!({ "FileSystemId": fsid, "Policy": policy }),
1366 ),
1367 None => Err(AwsServiceError::aws_error(
1368 StatusCode::NOT_FOUND,
1369 "PolicyNotFound",
1370 format!("No policy is attached to file system '{fsid}'."),
1371 )),
1372 }
1373 }
1374
1375 fn delete_file_system_policy(
1376 &self,
1377 ctx: &Ctx,
1378 label: &str,
1379 ) -> Result<AwsResponse, AwsServiceError> {
1380 let fsid = normalize_fs_id(label);
1381 let mut guard = self.state.write();
1382 let data = self.account(&mut guard, ctx);
1383 if !data.file_systems.contains_key(&fsid) {
1384 return Err(fs_not_found(&fsid));
1385 }
1386 data.file_system_policies.remove(&fsid);
1387 empty(StatusCode::OK)
1388 }
1389}
1390
1391impl EfsService {
1394 fn create_replication_configuration(
1395 &self,
1396 ctx: &Ctx,
1397 label: &str,
1398 b: &Value,
1399 ) -> Result<AwsResponse, AwsServiceError> {
1400 let fsid = normalize_fs_id(label);
1401 let mut guard = self.state.write();
1402 let data = self.account(&mut guard, ctx);
1403 if !data.file_systems.contains_key(&fsid) {
1404 return Err(fs_not_found(&fsid));
1405 }
1406 let destinations: Vec<Value> = b
1407 .get("Destinations")
1408 .and_then(Value::as_array)
1409 .cloned()
1410 .unwrap_or_default()
1411 .iter()
1412 .map(|d| {
1413 let region = d
1414 .get("Region")
1415 .and_then(Value::as_str)
1416 .unwrap_or(&ctx.region)
1417 .to_string();
1418 let dest_fs = d
1419 .get("FileSystemId")
1420 .and_then(Value::as_str)
1421 .map(normalize_fs_id)
1422 .unwrap_or_else(|| format!("fs-{}", hex17()));
1423 let mut dest = Map::new();
1424 dest.insert("Status".into(), json!("ENABLED"));
1425 dest.insert("FileSystemId".into(), json!(dest_fs));
1426 dest.insert("Region".into(), json!(region));
1427 dest.insert("LastReplicatedTimestamp".into(), json!(now_ts()));
1428 dest.insert("OwnerId".into(), json!(ctx.account));
1429 if let Some(role) = d.get("RoleArn") {
1430 dest.insert("RoleArn".into(), role.clone());
1431 }
1432 Value::Object(dest)
1433 })
1434 .collect();
1435
1436 let desc = json!({
1437 "SourceFileSystemId": fsid,
1438 "SourceFileSystemRegion": ctx.region,
1439 "SourceFileSystemArn": fs_arn(ctx, &fsid),
1440 "OriginalSourceFileSystemArn": fs_arn(ctx, &fsid),
1441 "CreationTime": now_ts(),
1442 "Destinations": destinations,
1443 "SourceFileSystemOwnerId": ctx.account,
1444 });
1445 data.replications.insert(fsid, desc.clone());
1446 ok(StatusCode::OK, desc)
1447 }
1448
1449 fn describe_replication_configurations(
1450 &self,
1451 ctx: &Ctx,
1452 q: &[(String, String)],
1453 ) -> Result<AwsResponse, AwsServiceError> {
1454 let by_fs = query_one(q, "FileSystemId").map(normalize_fs_id);
1455 let guard = self.state.read();
1456 let data = guard.get(&ctx.account);
1457 if let Some(fsid) = &by_fs {
1458 let fs_exists = data
1459 .map(|d| d.file_systems.contains_key(fsid))
1460 .unwrap_or(false);
1461 let has_replication = data
1462 .map(|d| d.replications.contains_key(fsid))
1463 .unwrap_or(false);
1464 if !fs_exists && !has_replication {
1465 return Err(fs_not_found(fsid));
1466 }
1467 if !has_replication {
1470 return Err(replication_not_found(fsid));
1471 }
1472 }
1473 let rows: Vec<Value> = data
1474 .map(|d| {
1475 d.replications
1476 .iter()
1477 .filter(|(src, _)| by_fs.as_deref().map(|f| src.as_str() == f).unwrap_or(true))
1478 .map(|(_, r)| r.clone())
1479 .collect()
1480 })
1481 .unwrap_or_default();
1482 ok(StatusCode::OK, json!({ "Replications": rows }))
1483 }
1484
1485 fn delete_replication_configuration(
1486 &self,
1487 ctx: &Ctx,
1488 label: &str,
1489 ) -> Result<AwsResponse, AwsServiceError> {
1490 let fsid = normalize_fs_id(label);
1491 let mut guard = self.state.write();
1492 let data = self.account(&mut guard, ctx);
1493 if !data.file_systems.contains_key(&fsid) {
1494 return Err(fs_not_found(&fsid));
1495 }
1496 if data.replications.remove(&fsid).is_none() {
1497 return Err(AwsServiceError::aws_error(
1498 StatusCode::NOT_FOUND,
1499 "ReplicationNotFound",
1500 format!("No replication configuration found for file system '{fsid}'."),
1501 ));
1502 }
1503 empty(StatusCode::NO_CONTENT)
1504 }
1505}
1506
1507impl EfsService {
1510 fn create_tags(
1511 &self,
1512 ctx: &Ctx,
1513 label: &str,
1514 b: &Value,
1515 ) -> Result<AwsResponse, AwsServiceError> {
1516 let fsid = normalize_fs_id(label);
1517 let mut guard = self.state.write();
1518 let data = self.account(&mut guard, ctx);
1519 if !data.file_systems.contains_key(&fsid) {
1520 return Err(fs_not_found(&fsid));
1521 }
1522 merge_tags(data, &fsid, b);
1523 sync_resource_tags(data, &fsid);
1524 empty(StatusCode::NO_CONTENT)
1525 }
1526
1527 fn delete_tags(
1528 &self,
1529 ctx: &Ctx,
1530 label: &str,
1531 b: &Value,
1532 ) -> Result<AwsResponse, AwsServiceError> {
1533 let fsid = normalize_fs_id(label);
1534 let mut guard = self.state.write();
1535 let data = self.account(&mut guard, ctx);
1536 if !data.file_systems.contains_key(&fsid) {
1537 return Err(fs_not_found(&fsid));
1538 }
1539 if let Some(keys) = b.get("TagKeys").and_then(Value::as_array) {
1540 if let Some(entry) = data.tags.get_mut(&fsid) {
1541 for k in keys {
1542 if let Some(ks) = k.as_str() {
1543 entry.remove(ks);
1544 }
1545 }
1546 }
1547 }
1548 sync_resource_tags(data, &fsid);
1549 empty(StatusCode::NO_CONTENT)
1550 }
1551
1552 fn describe_tags(
1553 &self,
1554 ctx: &Ctx,
1555 label: &str,
1556 q: &[(String, String)],
1557 ) -> Result<AwsResponse, AwsServiceError> {
1558 let fsid = normalize_fs_id(label);
1559 let guard = self.state.read();
1560 let data = guard.get(&ctx.account);
1561 let exists = data
1562 .map(|d| d.file_systems.contains_key(&fsid))
1563 .unwrap_or(false);
1564 if !exists {
1565 return Err(fs_not_found(&fsid));
1566 }
1567 let all = data
1568 .map(|d| tags_list(d, &fsid))
1569 .unwrap_or_else(|| json!([]));
1570 let rows = all.as_array().cloned().unwrap_or_default();
1571 let (page, next) =
1573 paginate_marker(rows, query_one(q, "Marker"), query_one(q, "MaxItems"), 100);
1574 let mut out = Map::new();
1575 out.insert("Tags".into(), Value::Array(page));
1576 if let Some(n) = next {
1577 out.insert("NextMarker".into(), json!(n));
1578 }
1579 ok(StatusCode::OK, Value::Object(out))
1580 }
1581
1582 fn tag_resource(
1583 &self,
1584 ctx: &Ctx,
1585 label: &str,
1586 b: &Value,
1587 ) -> Result<AwsResponse, AwsServiceError> {
1588 let rid = normalize_fs_id(label);
1589 let mut guard = self.state.write();
1590 let data = self.account(&mut guard, ctx);
1591 self.require_taggable(data, &rid)?;
1592 merge_tags(data, &rid, b);
1593 sync_resource_tags(data, &rid);
1594 empty(StatusCode::OK)
1595 }
1596
1597 fn untag_resource(
1598 &self,
1599 ctx: &Ctx,
1600 label: &str,
1601 q: &[(String, String)],
1602 ) -> Result<AwsResponse, AwsServiceError> {
1603 let rid = normalize_fs_id(label);
1604 let mut guard = self.state.write();
1605 let data = self.account(&mut guard, ctx);
1606 self.require_taggable(data, &rid)?;
1607 let keys = query_all(q, "tagKeys");
1608 if let Some(entry) = data.tags.get_mut(&rid) {
1609 for k in &keys {
1610 entry.remove(k);
1611 }
1612 }
1613 sync_resource_tags(data, &rid);
1614 empty(StatusCode::OK)
1615 }
1616
1617 fn list_tags_for_resource(
1618 &self,
1619 ctx: &Ctx,
1620 label: &str,
1621 q: &[(String, String)],
1622 ) -> Result<AwsResponse, AwsServiceError> {
1623 let rid = normalize_fs_id(label);
1624 let guard = self.state.read();
1625 let data = guard.get(&ctx.account);
1626 let known = data
1628 .map(|d| {
1629 (rid.starts_with("fsap-") && d.access_points.contains_key(&rid))
1630 || (rid.starts_with("fs-") && d.file_systems.contains_key(&rid))
1631 })
1632 .unwrap_or(false);
1633 if !known {
1634 return Err(if rid.starts_with("fsap-") {
1635 ap_not_found(&rid)
1636 } else {
1637 fs_not_found(&rid)
1638 });
1639 }
1640 let all = data
1641 .map(|d| tags_list(d, &rid))
1642 .unwrap_or_else(|| json!([]));
1643 let rows = all.as_array().cloned().unwrap_or_default();
1644 let start = query_one(q, "NextToken")
1646 .and_then(|t| t.parse::<usize>().ok())
1647 .unwrap_or(0);
1648 let max = query_one(q, "MaxResults")
1649 .and_then(|m| m.parse::<usize>().ok())
1650 .map(|m| m.max(1))
1651 .unwrap_or(100);
1652 let end = start.saturating_add(max).min(rows.len());
1653 let page = rows.get(start..end).unwrap_or(&[]).to_vec();
1654 let mut out = Map::new();
1655 out.insert("Tags".into(), Value::Array(page));
1656 if end < rows.len() {
1657 out.insert("NextToken".into(), json!(end.to_string()));
1658 }
1659 ok(StatusCode::OK, Value::Object(out))
1660 }
1661
1662 fn require_taggable(&self, data: &EfsData, rid: &str) -> Result<(), AwsServiceError> {
1663 if rid.starts_with("fsap-") {
1664 if data.access_points.contains_key(rid) {
1665 Ok(())
1666 } else {
1667 Err(ap_not_found(rid))
1668 }
1669 } else if data.file_systems.contains_key(rid) {
1670 Ok(())
1671 } else {
1672 Err(fs_not_found(rid))
1673 }
1674 }
1675
1676 fn describe_account_preferences(&self, ctx: &Ctx) -> Result<AwsResponse, AwsServiceError> {
1677 let guard = self.state.read();
1678 match guard
1681 .get(&ctx.account)
1682 .and_then(|d| d.resource_id_preference.clone())
1683 {
1684 Some(pref) => ok(
1685 StatusCode::OK,
1686 json!({
1687 "ResourceIdPreference": {
1688 "ResourceIdType": pref,
1689 "Resources": ["FILE_SYSTEM", "MOUNT_TARGET"]
1690 }
1691 }),
1692 ),
1693 None => empty(StatusCode::OK),
1694 }
1695 }
1696
1697 fn put_account_preferences(
1698 &self,
1699 ctx: &Ctx,
1700 b: &Value,
1701 ) -> Result<AwsResponse, AwsServiceError> {
1702 let id_type = b
1703 .get("ResourceIdType")
1704 .and_then(Value::as_str)
1705 .unwrap_or("LONG_ID")
1706 .to_string();
1707 let mut guard = self.state.write();
1708 let data = self.account(&mut guard, ctx);
1709 data.resource_id_preference = Some(id_type.clone());
1710 ok(
1711 StatusCode::OK,
1712 json!({
1713 "ResourceIdPreference": {
1714 "ResourceIdType": id_type,
1715 "Resources": ["FILE_SYSTEM", "MOUNT_TARGET"]
1716 }
1717 }),
1718 )
1719 }
1720}
1721
1722fn merge_tags(data: &mut EfsData, resource_id: &str, b: &Value) {
1724 let map = tags_to_map(b);
1725 if map.is_empty() {
1726 return;
1727 }
1728 let entry = data.tags.entry(resource_id.to_string()).or_default();
1729 for (k, v) in &map {
1730 if let Some(vs) = v.as_str() {
1731 entry.insert(k.clone(), vs.to_string());
1732 }
1733 }
1734}
1735
1736fn sync_resource_tags(data: &mut EfsData, rid: &str) {
1741 let tags = data.tags.get(rid).cloned().unwrap_or_default();
1742 let obj = if rid.starts_with("fsap-") {
1743 data.access_points.get_mut(rid)
1744 } else {
1745 data.file_systems.get_mut(rid)
1746 }
1747 .and_then(Value::as_object_mut);
1748 if let Some(obj) = obj {
1749 match tags.get("Name") {
1750 Some(name) => {
1751 obj.insert("Name".into(), json!(name));
1752 }
1753 None => {
1754 obj.remove("Name");
1755 }
1756 }
1757 obj.insert(
1758 "Tags".into(),
1759 Value::Array(
1760 tags.iter()
1761 .map(|(k, v)| json!({ "Key": k, "Value": v }))
1762 .collect(),
1763 ),
1764 );
1765 }
1766}
1767
1768#[cfg(test)]
1769mod tests {
1770 use super::*;
1771 use parking_lot::RwLock;
1772
1773 fn ctx() -> Ctx {
1774 Ctx {
1775 account: "000000000000".to_string(),
1776 region: "us-east-1".to_string(),
1777 }
1778 }
1779
1780 fn svc() -> EfsService {
1781 let state: SharedEfsState = Arc::new(RwLock::new(MultiAccountState::new(
1782 "000000000000",
1783 "us-east-1",
1784 "",
1785 )));
1786 EfsService::new(state)
1787 }
1788
1789 fn empty_ec2() -> fakecloud_ec2::SharedEc2State {
1790 Arc::new(RwLock::new(MultiAccountState::new(
1791 "000000000000",
1792 "us-east-1",
1793 "",
1794 )))
1795 }
1796
1797 fn seed_fs(s: &EfsService, fsid: &str, life_cycle: &str) {
1798 let mut g = s.state.write();
1799 let d = g.get_or_create("000000000000");
1800 d.file_systems.insert(
1801 fsid.to_string(),
1802 json!({ "FileSystemId": fsid, "LifeCycleState": life_cycle }),
1803 );
1804 }
1805
1806 fn body_value(resp: &AwsResponse) -> Value {
1807 serde_json::from_slice(resp.body.expect_bytes()).unwrap()
1808 }
1809
1810 #[test]
1813 fn create_mount_target_nonexistent_subnet_is_subnet_not_found() {
1814 let s = svc().with_ec2_state(empty_ec2());
1815 seed_fs(&s, "fs-1", "available");
1816 let err = s
1817 .create_mount_target(
1818 &ctx(),
1819 &json!({ "FileSystemId": "fs-1", "SubnetId": "subnet-does-not-exist" }),
1820 )
1821 .err()
1822 .unwrap();
1823 assert_eq!(err.code(), "SubnetNotFound");
1824 assert_eq!(err.status(), StatusCode::BAD_REQUEST);
1825 }
1826
1827 #[test]
1830 fn create_mount_target_without_ec2_synthesizes_subnet() {
1831 let s = svc();
1832 seed_fs(&s, "fs-1", "available");
1833 let resp = s
1834 .create_mount_target(
1835 &ctx(),
1836 &json!({ "FileSystemId": "fs-1", "SubnetId": "subnet-synth" }),
1837 )
1838 .unwrap();
1839 assert!(resp.status.is_success());
1840 }
1841
1842 #[test]
1845 fn create_mount_target_on_creating_fs_is_incorrect_lifecycle() {
1846 let s = svc();
1847 seed_fs(&s, "fs-1", "creating");
1848 let err = s
1849 .create_mount_target(
1850 &ctx(),
1851 &json!({ "FileSystemId": "fs-1", "SubnetId": "subnet-1" }),
1852 )
1853 .err()
1854 .unwrap();
1855 assert_eq!(err.code(), "IncorrectFileSystemLifeCycleState");
1856 assert_eq!(err.status(), StatusCode::CONFLICT);
1857 }
1858
1859 #[test]
1861 fn delete_file_system_with_access_point_is_in_use() {
1862 let s = svc();
1863 seed_fs(&s, "fs-1", "available");
1864 {
1865 let mut g = s.state.write();
1866 let d = g.get_or_create("000000000000");
1867 d.access_points.insert(
1868 "fsap-1".to_string(),
1869 json!({ "AccessPointId": "fsap-1", "FileSystemId": "fs-1" }),
1870 );
1871 }
1872 let err = s.delete_file_system(&ctx(), "fs-1").err().unwrap();
1873 assert_eq!(err.code(), "FileSystemInUse");
1874 assert_eq!(err.status(), StatusCode::CONFLICT);
1875 }
1876
1877 #[test]
1880 fn describe_replication_missing_is_replication_not_found() {
1881 let s = svc();
1882 seed_fs(&s, "fs-1", "available");
1883 let err = s
1884 .describe_replication_configurations(
1885 &ctx(),
1886 &[("FileSystemId".to_string(), "fs-1".to_string())],
1887 )
1888 .err()
1889 .unwrap();
1890 assert_eq!(err.code(), "ReplicationNotFound");
1891 assert_eq!(err.status(), StatusCode::NOT_FOUND);
1892 }
1893
1894 #[test]
1896 fn describe_replication_no_filter_is_empty_ok() {
1897 let s = svc();
1898 let resp = s.describe_replication_configurations(&ctx(), &[]).unwrap();
1899 assert!(resp.status.is_success());
1900 assert_eq!(
1901 body_value(&resp)["Replications"].as_array().unwrap().len(),
1902 0
1903 );
1904 }
1905
1906 #[test]
1909 fn describe_mount_targets_two_filters_is_bad_request() {
1910 let s = svc();
1911 let err = s
1912 .describe_mount_targets(
1913 &ctx(),
1914 &[
1915 ("FileSystemId".to_string(), "fs-1".to_string()),
1916 ("MountTargetId".to_string(), "fsmt-1".to_string()),
1917 ],
1918 )
1919 .err()
1920 .unwrap();
1921 assert_eq!(err.code(), "BadRequest");
1922 assert_eq!(err.status(), StatusCode::BAD_REQUEST);
1923 }
1924
1925 #[test]
1928 fn create_file_system_provisioned_without_mibps_is_bad_request() {
1929 let s = svc();
1930 let err = s
1931 .create_file_system(
1932 &ctx(),
1933 &json!({ "CreationToken": "tok-1", "ThroughputMode": "provisioned" }),
1934 )
1935 .err()
1936 .unwrap();
1937 assert_eq!(err.code(), "BadRequest");
1938
1939 let err2 = s
1940 .create_file_system(
1941 &ctx(),
1942 &json!({
1943 "CreationToken": "tok-2",
1944 "ThroughputMode": "bursting",
1945 "ProvisionedThroughputInMibps": 128
1946 }),
1947 )
1948 .err()
1949 .unwrap();
1950 assert_eq!(err2.code(), "BadRequest");
1951
1952 let ok = s
1954 .create_file_system(
1955 &ctx(),
1956 &json!({
1957 "CreationToken": "tok-3",
1958 "ThroughputMode": "provisioned",
1959 "ProvisionedThroughputInMibps": 256
1960 }),
1961 )
1962 .unwrap();
1963 assert!(ok.status.is_success());
1964 }
1965
1966 #[test]
1969 fn describe_account_preferences_fresh_account_is_empty() {
1970 let s = svc();
1971 let resp = s.describe_account_preferences(&ctx()).unwrap();
1972 assert!(resp.status.is_success());
1973 let v = body_value(&resp);
1974 assert!(
1975 v.get("ResourceIdPreference").is_none(),
1976 "expected empty {{}}, got {v}"
1977 );
1978
1979 s.put_account_preferences(&ctx(), &json!({ "ResourceIdType": "SHORT_ID" }))
1981 .unwrap();
1982 let resp2 = s.describe_account_preferences(&ctx()).unwrap();
1983 let v2 = body_value(&resp2);
1984 assert_eq!(v2["ResourceIdPreference"]["ResourceIdType"], "SHORT_ID");
1985 }
1986
1987 #[test]
1990 fn access_point_tags_resync_agrees() {
1991 let s = svc();
1992 seed_fs(&s, "fs-1", "available");
1993 let created = s
1994 .create_access_point(
1995 &ctx(),
1996 &json!({ "ClientToken": "ct-1", "FileSystemId": "fs-1" }),
1997 )
1998 .unwrap();
1999 let apid = body_value(&created)["AccessPointId"]
2000 .as_str()
2001 .unwrap()
2002 .to_string();
2003
2004 s.tag_resource(
2005 &ctx(),
2006 &apid,
2007 &json!({ "Tags": [{ "Key": "Env", "Value": "prod" }] }),
2008 )
2009 .unwrap();
2010
2011 let embedded: Vec<(String, String)> = {
2013 let g = s.state.read();
2014 let ap = g
2015 .get("000000000000")
2016 .unwrap()
2017 .access_points
2018 .get(&apid)
2019 .unwrap()
2020 .clone();
2021 ap["Tags"]
2022 .as_array()
2023 .unwrap()
2024 .iter()
2025 .map(|t| {
2026 (
2027 t["Key"].as_str().unwrap().to_string(),
2028 t["Value"].as_str().unwrap().to_string(),
2029 )
2030 })
2031 .collect()
2032 };
2033
2034 let listed_resp = s.list_tags_for_resource(&ctx(), &apid, &[]).unwrap();
2036 let listed: Vec<(String, String)> = body_value(&listed_resp)["Tags"]
2037 .as_array()
2038 .unwrap()
2039 .iter()
2040 .map(|t| {
2041 (
2042 t["Key"].as_str().unwrap().to_string(),
2043 t["Value"].as_str().unwrap().to_string(),
2044 )
2045 })
2046 .collect();
2047
2048 assert_eq!(embedded, vec![("Env".to_string(), "prod".to_string())]);
2049 assert_eq!(
2050 embedded, listed,
2051 "embedded Tags must match ListTagsForResource"
2052 );
2053
2054 s.untag_resource(&ctx(), &apid, &[("tagKeys".to_string(), "Env".to_string())])
2056 .unwrap();
2057 let g = s.state.read();
2058 let ap = g
2059 .get("000000000000")
2060 .unwrap()
2061 .access_points
2062 .get(&apid)
2063 .unwrap();
2064 assert_eq!(ap["Tags"].as_array().unwrap().len(), 0);
2065 }
2066}