use std::sync::Arc;
use async_trait::async_trait;
use http::{Method, StatusCode};
use parking_lot::RwLockWriteGuard;
use percent_encoding::percent_decode_str;
use serde_json::{json, Map, Value};
use tokio::sync::Mutex as AsyncMutex;
use uuid::Uuid;
use fakecloud_core::multi_account::MultiAccountState;
use fakecloud_core::service::{AwsRequest, AwsResponse, AwsService, AwsServiceError};
use fakecloud_persistence::SnapshotStore;
use crate::persistence::save_snapshot;
use crate::state::{EfsData, SharedEfsState};
pub const EFS_ACTIONS: &[&str] = &[
"CreateAccessPoint",
"CreateFileSystem",
"CreateMountTarget",
"CreateReplicationConfiguration",
"CreateTags",
"DeleteAccessPoint",
"DeleteFileSystem",
"DeleteFileSystemPolicy",
"DeleteMountTarget",
"DeleteReplicationConfiguration",
"DeleteTags",
"DescribeAccessPoints",
"DescribeAccountPreferences",
"DescribeBackupPolicy",
"DescribeFileSystemPolicy",
"DescribeFileSystems",
"DescribeLifecycleConfiguration",
"DescribeMountTargetSecurityGroups",
"DescribeMountTargets",
"DescribeReplicationConfigurations",
"DescribeTags",
"ListTagsForResource",
"ModifyMountTargetSecurityGroups",
"PutAccountPreferences",
"PutBackupPolicy",
"PutFileSystemPolicy",
"PutLifecycleConfiguration",
"TagResource",
"UntagResource",
"UpdateFileSystem",
"UpdateFileSystemProtection",
];
const MUTATING: &[&str] = &[
"CreateAccessPoint",
"CreateFileSystem",
"CreateMountTarget",
"CreateReplicationConfiguration",
"CreateTags",
"DeleteAccessPoint",
"DeleteFileSystem",
"DeleteFileSystemPolicy",
"DeleteMountTarget",
"DeleteReplicationConfiguration",
"DeleteTags",
"ModifyMountTargetSecurityGroups",
"PutAccountPreferences",
"PutBackupPolicy",
"PutFileSystemPolicy",
"PutLifecycleConfiguration",
"TagResource",
"UntagResource",
"UpdateFileSystem",
"UpdateFileSystemProtection",
];
pub struct EfsService {
state: SharedEfsState,
snapshot_store: Option<Arc<dyn SnapshotStore>>,
snapshot_lock: Arc<AsyncMutex<()>>,
ec2_state: Option<fakecloud_ec2::SharedEc2State>,
}
impl EfsService {
pub fn new(state: SharedEfsState) -> Self {
Self {
state,
snapshot_store: None,
snapshot_lock: Arc::new(AsyncMutex::new(())),
ec2_state: None,
}
}
pub fn with_snapshot_store(mut self, store: Arc<dyn SnapshotStore>) -> Self {
self.snapshot_store = Some(store);
self
}
pub fn with_ec2_state(mut self, ec2_state: fakecloud_ec2::SharedEc2State) -> Self {
self.ec2_state = Some(ec2_state);
self
}
pub fn snapshot_hook(&self) -> Option<fakecloud_persistence::SnapshotHook> {
let store = self.snapshot_store.clone()?;
let state = self.state.clone();
let lock = self.snapshot_lock.clone();
Some(Arc::new(move || {
let state = state.clone();
let store = store.clone();
let lock = lock.clone();
Box::pin(async move {
save_snapshot(&state, Some(store), &lock).await;
})
}))
}
fn resolve_subnet(&self, account: &str, subnet_id: &str) -> Option<(String, String, String)> {
let ec2 = self.ec2_state.as_ref()?;
let guard = ec2.read();
let subnet = guard.get(account)?.subnets.get(subnet_id)?;
Some((
subnet.availability_zone.clone(),
subnet.availability_zone_id.clone(),
subnet.vpc_id.clone(),
))
}
async fn save(&self) {
save_snapshot(
&self.state,
self.snapshot_store.clone(),
&self.snapshot_lock,
)
.await;
}
fn resolve_action(req: &AwsRequest) -> Option<(&'static str, Option<String>)> {
let raw = req.raw_path.split('?').next().unwrap_or(&req.raw_path);
let trimmed = raw.strip_prefix('/').unwrap_or(raw);
let segs: Vec<String> = if trimmed.is_empty() {
Vec::new()
} else {
trimmed
.split('/')
.map(|s| percent_decode_str(s).decode_utf8_lossy().into_owned())
.collect()
};
let s: Vec<&str> = segs.iter().map(String::as_str).collect();
let m = &req.method;
let get = m == Method::GET;
let post = m == Method::POST;
let put = m == Method::PUT;
let del = m == Method::DELETE;
const V: &str = "2015-02-01";
let (action, label): (&'static str, Option<String>) = match s.as_slice() {
[v, "access-points"] if *v == V && post => ("CreateAccessPoint", None),
[v, "access-points"] if *v == V && get => ("DescribeAccessPoints", None),
[v, "access-points", id] if *v == V && del => {
("DeleteAccessPoint", Some((*id).to_string()))
}
[v, "mount-targets"] if *v == V && post => ("CreateMountTarget", None),
[v, "mount-targets"] if *v == V && get => ("DescribeMountTargets", None),
[v, "mount-targets", id] if *v == V && del => {
("DeleteMountTarget", Some((*id).to_string()))
}
[v, "mount-targets", id, "security-groups"] if *v == V && get => {
("DescribeMountTargetSecurityGroups", Some((*id).to_string()))
}
[v, "mount-targets", id, "security-groups"] if *v == V && put => {
("ModifyMountTargetSecurityGroups", Some((*id).to_string()))
}
[v, "account-preferences"] if *v == V && get => ("DescribeAccountPreferences", None),
[v, "account-preferences"] if *v == V && put => ("PutAccountPreferences", None),
[v, "file-systems"] if *v == V && post => ("CreateFileSystem", None),
[v, "file-systems"] if *v == V && get => ("DescribeFileSystems", None),
[v, "file-systems", "replication-configurations"] if *v == V && get => {
("DescribeReplicationConfigurations", None)
}
[v, "file-systems", id, "replication-configuration"] if *v == V && post => {
("CreateReplicationConfiguration", Some((*id).to_string()))
}
[v, "file-systems", id, "replication-configuration"] if *v == V && del => {
("DeleteReplicationConfiguration", Some((*id).to_string()))
}
[v, "file-systems", id, "policy"] if *v == V && get => {
("DescribeFileSystemPolicy", Some((*id).to_string()))
}
[v, "file-systems", id, "policy"] if *v == V && put => {
("PutFileSystemPolicy", Some((*id).to_string()))
}
[v, "file-systems", id, "policy"] if *v == V && del => {
("DeleteFileSystemPolicy", Some((*id).to_string()))
}
[v, "file-systems", id, "backup-policy"] if *v == V && get => {
("DescribeBackupPolicy", Some((*id).to_string()))
}
[v, "file-systems", id, "backup-policy"] if *v == V && put => {
("PutBackupPolicy", Some((*id).to_string()))
}
[v, "file-systems", id, "lifecycle-configuration"] if *v == V && get => {
("DescribeLifecycleConfiguration", Some((*id).to_string()))
}
[v, "file-systems", id, "lifecycle-configuration"] if *v == V && put => {
("PutLifecycleConfiguration", Some((*id).to_string()))
}
[v, "file-systems", id, "protection"] if *v == V && put => {
("UpdateFileSystemProtection", Some((*id).to_string()))
}
[v, "file-systems", id] if *v == V && del => {
("DeleteFileSystem", Some((*id).to_string()))
}
[v, "file-systems", id] if *v == V && put => {
("UpdateFileSystem", Some((*id).to_string()))
}
[v, "create-tags", id] if *v == V && post => ("CreateTags", Some((*id).to_string())),
[v, "delete-tags", id] if *v == V && post => ("DeleteTags", Some((*id).to_string())),
[v, "tags", id] if *v == V && get => ("DescribeTags", Some((*id).to_string())),
[v, "resource-tags", id] if *v == V && get => {
("ListTagsForResource", Some((*id).to_string()))
}
[v, "resource-tags", id] if *v == V && post => ("TagResource", Some((*id).to_string())),
[v, "resource-tags", id] if *v == V && del => {
("UntagResource", Some((*id).to_string()))
}
_ => return None,
};
Some((action, label))
}
}
#[async_trait]
impl AwsService for EfsService {
fn service_name(&self) -> &str {
"elasticfilesystem"
}
async fn handle(&self, req: AwsRequest) -> Result<AwsResponse, AwsServiceError> {
let Some((action, label)) = Self::resolve_action(&req) else {
return Err(AwsServiceError::aws_error(
StatusCode::NOT_FOUND,
"UnknownOperationException",
format!("Unknown operation: {} {}", req.method, req.raw_path),
));
};
let result = self.dispatch(action, label, &req);
if MUTATING.contains(&action)
&& matches!(result.as_ref(), Ok(resp) if resp.status.is_success())
{
self.save().await;
}
result
}
fn supported_actions(&self) -> &[&str] {
EFS_ACTIONS
}
}
struct Ctx {
account: String,
region: String,
}
impl EfsService {
#[allow(clippy::too_many_lines)]
fn dispatch(
&self,
action: &str,
label: Option<String>,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let body = parse_body(req)?;
crate::validate::validate_input(action, &body)?;
let ctx = Ctx {
account: req.account_id.clone(),
region: req.region.clone(),
};
let q = parse_query(&req.raw_query);
crate::validate::validate_query(&q)?;
let label = label.unwrap_or_default();
match action {
"CreateFileSystem" => self.create_file_system(&ctx, &body),
"DescribeFileSystems" => self.describe_file_systems(&ctx, &q),
"DeleteFileSystem" => self.delete_file_system(&ctx, &label),
"UpdateFileSystem" => self.update_file_system(&ctx, &label, &body),
"UpdateFileSystemProtection" => self.update_file_system_protection(&ctx, &label, &body),
"CreateMountTarget" => self.create_mount_target(&ctx, &body),
"DescribeMountTargets" => self.describe_mount_targets(&ctx, &q),
"DeleteMountTarget" => self.delete_mount_target(&ctx, &label),
"DescribeMountTargetSecurityGroups" => self.describe_mt_security_groups(&ctx, &label),
"ModifyMountTargetSecurityGroups" => {
self.modify_mt_security_groups(&ctx, &label, &body)
}
"CreateAccessPoint" => self.create_access_point(&ctx, &body),
"DescribeAccessPoints" => self.describe_access_points(&ctx, &q),
"DeleteAccessPoint" => self.delete_access_point(&ctx, &label),
"PutLifecycleConfiguration" => self.put_lifecycle_configuration(&ctx, &label, &body),
"DescribeLifecycleConfiguration" => self.describe_lifecycle_configuration(&ctx, &label),
"PutBackupPolicy" => self.put_backup_policy(&ctx, &label, &body),
"DescribeBackupPolicy" => self.describe_backup_policy(&ctx, &label),
"PutFileSystemPolicy" => self.put_file_system_policy(&ctx, &label, &body),
"DescribeFileSystemPolicy" => self.describe_file_system_policy(&ctx, &label),
"DeleteFileSystemPolicy" => self.delete_file_system_policy(&ctx, &label),
"CreateReplicationConfiguration" => {
self.create_replication_configuration(&ctx, &label, &body)
}
"DescribeReplicationConfigurations" => {
self.describe_replication_configurations(&ctx, &q)
}
"DeleteReplicationConfiguration" => self.delete_replication_configuration(&ctx, &label),
"CreateTags" => self.create_tags(&ctx, &label, &body),
"DeleteTags" => self.delete_tags(&ctx, &label, &body),
"DescribeTags" => self.describe_tags(&ctx, &label, &q),
"TagResource" => self.tag_resource(&ctx, &label, &body),
"UntagResource" => self.untag_resource(&ctx, &label, &q),
"ListTagsForResource" => self.list_tags_for_resource(&ctx, &label, &q),
"DescribeAccountPreferences" => self.describe_account_preferences(&ctx),
"PutAccountPreferences" => self.put_account_preferences(&ctx, &body),
_ => Err(AwsServiceError::action_not_implemented(
"elasticfilesystem",
action,
)),
}
}
}
fn ok(status: StatusCode, v: Value) -> Result<AwsResponse, AwsServiceError> {
Ok(AwsResponse::json_value(status, v))
}
fn empty(status: StatusCode) -> Result<AwsResponse, AwsServiceError> {
Ok(AwsResponse::json_value(status, json!({})))
}
fn parse_body(req: &AwsRequest) -> Result<Value, AwsServiceError> {
if req.body.is_empty() {
return Ok(json!({}));
}
serde_json::from_slice(&req.body)
.map_err(|e| bad_request(&format!("Request body is malformed: {e}")))
}
fn bad_request(msg: &str) -> AwsServiceError {
AwsServiceError::aws_error(StatusCode::BAD_REQUEST, "BadRequest", msg)
}
fn fs_not_found(id: &str) -> AwsServiceError {
AwsServiceError::aws_error(
StatusCode::NOT_FOUND,
"FileSystemNotFound",
format!("File system '{id}' does not exist."),
)
}
fn mt_not_found(id: &str) -> AwsServiceError {
AwsServiceError::aws_error(
StatusCode::NOT_FOUND,
"MountTargetNotFound",
format!("Mount target '{id}' does not exist."),
)
}
fn ap_not_found(id: &str) -> AwsServiceError {
AwsServiceError::aws_error(
StatusCode::NOT_FOUND,
"AccessPointNotFound",
format!("Access point '{id}' does not exist."),
)
}
fn subnet_not_found(id: &str) -> AwsServiceError {
AwsServiceError::aws_error(
StatusCode::BAD_REQUEST,
"SubnetNotFound",
format!("The subnet ID '{id}' is invalid or does not exist."),
)
}
fn replication_not_found(id: &str) -> AwsServiceError {
AwsServiceError::aws_error(
StatusCode::NOT_FOUND,
"ReplicationNotFound",
format!("Replication configuration for file system '{id}' does not exist."),
)
}
fn incorrect_fs_lifecycle_state(id: &str, state: &str) -> AwsServiceError {
AwsServiceError::aws_error(
StatusCode::CONFLICT,
"IncorrectFileSystemLifeCycleState",
format!(
"File system '{id}' is in life cycle state '{state}' but must be 'available' for this operation."
),
)
}
fn conflict(code: &str, msg: &str) -> AwsServiceError {
AwsServiceError::aws_error(StatusCode::CONFLICT, code, msg)
}
fn hex17() -> String {
Uuid::new_v4()
.simple()
.to_string()
.chars()
.filter(char::is_ascii_hexdigit)
.take(17)
.collect()
}
fn now_ts() -> f64 {
chrono::Utc::now().timestamp() as f64
}
fn fs_arn(ctx: &Ctx, fsid: &str) -> String {
format!(
"arn:aws:elasticfilesystem:{}:{}:file-system/{}",
ctx.region, ctx.account, fsid
)
}
fn ap_arn(ctx: &Ctx, apid: &str) -> String {
format!(
"arn:aws:elasticfilesystem:{}:{}:access-point/{}",
ctx.region, ctx.account, apid
)
}
fn normalize_fs_id(raw: &str) -> String {
raw.rsplit('/').next().unwrap_or(raw).to_string()
}
fn hash_str(s: &str) -> u64 {
let mut h: u64 = 0xcbf2_9ce4_8422_2325;
for b in s.as_bytes() {
h ^= *b as u64;
h = h.wrapping_mul(0x0000_0100_0000_01b3);
}
h
}
fn parse_query(raw: &str) -> Vec<(String, String)> {
raw.split('&')
.filter(|p| !p.is_empty())
.map(|pair| {
let (k, v) = pair.split_once('=').unwrap_or((pair, ""));
(
percent_decode_str(k).decode_utf8_lossy().into_owned(),
percent_decode_str(v).decode_utf8_lossy().into_owned(),
)
})
.collect()
}
fn query_one<'a>(q: &'a [(String, String)], key: &str) -> Option<&'a str> {
q.iter()
.find(|(k, _)| k == key)
.map(|(_, v)| v.as_str())
.filter(|v| !v.is_empty())
}
fn query_all(q: &[(String, String)], key: &str) -> Vec<String> {
q.iter()
.filter(|(k, _)| k == key)
.map(|(_, v)| v.clone())
.collect()
}
fn tags_to_map(b: &Value) -> Map<String, Value> {
let mut out = Map::new();
if let Some(arr) = b.get("Tags").and_then(Value::as_array) {
for t in arr {
if let (Some(k), Some(v)) = (
t.get("Key").and_then(Value::as_str),
t.get("Value").and_then(Value::as_str),
) {
out.insert(k.to_string(), json!(v));
}
}
}
out
}
fn tags_list(data: &EfsData, resource_id: &str) -> Value {
let arr: Vec<Value> = data
.tags
.get(resource_id)
.map(|m| {
m.iter()
.map(|(k, v)| json!({ "Key": k, "Value": v }))
.collect()
})
.unwrap_or_default();
Value::Array(arr)
}
impl EfsService {
fn account<'g>(
&self,
guard: &'g mut RwLockWriteGuard<'_, MultiAccountState<EfsData>>,
ctx: &Ctx,
) -> &'g mut EfsData {
guard.get_or_create(&ctx.account)
}
}
impl EfsService {
fn create_file_system(&self, ctx: &Ctx, b: &Value) -> Result<AwsResponse, AwsServiceError> {
let creation_token = b
.get("CreationToken")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string();
let mut guard = self.state.write();
let data = guard.get_or_create(&ctx.account);
if let Some((existing_id, _)) = data.file_systems.iter().find(|(_, fs)| {
fs.get("CreationToken").and_then(Value::as_str) == Some(creation_token.as_str())
}) {
let existing_id = existing_id.clone();
return Err(AwsServiceError::aws_error_with_fields(
StatusCode::CONFLICT,
"FileSystemAlreadyExists",
format!("File system already exists with creation token {creation_token}"),
vec![("FileSystemId".to_string(), existing_id)],
));
}
let fsid = format!("fs-{}", hex17());
let performance_mode = b
.get("PerformanceMode")
.and_then(Value::as_str)
.unwrap_or("generalPurpose");
let throughput_mode = b
.get("ThroughputMode")
.and_then(Value::as_str)
.unwrap_or("bursting");
if throughput_mode == "provisioned" {
match b
.get("ProvisionedThroughputInMibps")
.and_then(Value::as_f64)
{
None => {
return Err(bad_request(
"ProvisionedThroughputInMibps is required when ThroughputMode is set to provisioned.",
));
}
Some(v) if v < 1.0 => {
return Err(bad_request(
"Value at 'ProvisionedThroughputInMibps' failed to satisfy constraint: Member must have value greater than or equal to 1",
));
}
Some(_) => {}
}
} else if b.get("ProvisionedThroughputInMibps").is_some() {
return Err(bad_request(
"ProvisionedThroughputInMibps is only applicable when ThroughputMode is set to provisioned.",
));
}
let encrypted = b.get("Encrypted").and_then(Value::as_bool).unwrap_or(false);
let mut fs = Map::new();
fs.insert("OwnerId".into(), json!(ctx.account));
fs.insert("CreationToken".into(), json!(creation_token));
fs.insert("FileSystemId".into(), json!(fsid));
fs.insert("FileSystemArn".into(), json!(fs_arn(ctx, &fsid)));
fs.insert("CreationTime".into(), json!(now_ts()));
fs.insert("LifeCycleState".into(), json!("creating"));
fs.insert("NumberOfMountTargets".into(), json!(0));
fs.insert(
"SizeInBytes".into(),
json!({
"Value": 6144,
"Timestamp": now_ts(),
"ValueInIA": 0,
"ValueInStandard": 6144,
"ValueInArchive": 0
}),
);
fs.insert("PerformanceMode".into(), json!(performance_mode));
fs.insert("ThroughputMode".into(), json!(throughput_mode));
fs.insert("Encrypted".into(), json!(encrypted));
if encrypted {
let kms = b
.get("KmsKeyId")
.and_then(Value::as_str)
.map(str::to_string)
.unwrap_or_else(|| {
format!(
"arn:aws:kms:{}:{}:key/{}-{}-{}-{}-{}",
ctx.region,
ctx.account,
&hex17()[..8],
&hex17()[..4],
&hex17()[..4],
&hex17()[..4],
&hex17()[..12.min(hex17().len())]
)
});
fs.insert("KmsKeyId".into(), json!(kms));
}
if throughput_mode == "provisioned" {
if let Some(p) = b.get("ProvisionedThroughputInMibps") {
fs.insert("ProvisionedThroughputInMibps".into(), p.clone());
}
}
if let Some(az) = b.get("AvailabilityZoneName").and_then(Value::as_str) {
fs.insert("AvailabilityZoneName".into(), json!(az));
fs.insert(
"AvailabilityZoneId".into(),
json!(format!("{}-az1", ctx.region)),
);
}
let tag_map = tags_to_map(b);
if let Some(name) = tag_map.get("Name").and_then(Value::as_str) {
fs.insert("Name".into(), json!(name));
}
fs.insert("Tags".into(), tags_list_from_map(&tag_map));
fs.insert(
"FileSystemProtection".into(),
json!({ "ReplicationOverwriteProtection": "ENABLED" }),
);
if !tag_map.is_empty() {
let entry = data.tags.entry(fsid.clone()).or_default();
for (k, v) in &tag_map {
if let Some(vs) = v.as_str() {
entry.insert(k.clone(), vs.to_string());
}
}
}
data.file_systems
.insert(fsid.clone(), Value::Object(fs.clone()));
ok(StatusCode::CREATED, Value::Object(fs))
}
fn describe_file_systems(
&self,
ctx: &Ctx,
q: &[(String, String)],
) -> Result<AwsResponse, AwsServiceError> {
let filter_id = query_one(q, "FileSystemId").map(normalize_fs_id);
let filter_token = query_one(q, "CreationToken");
let mut guard = self.state.write();
let data = self.account(&mut guard, ctx);
data.reconcile_lifecycle();
if let Some(id) = &filter_id {
if !data.file_systems.contains_key(id) {
return Err(fs_not_found(id));
}
}
let rows: Vec<Value> = data
.file_systems
.values()
.filter(|fs| {
filter_id
.as_ref()
.map(|id| fs.get("FileSystemId").and_then(Value::as_str) == Some(id.as_str()))
.unwrap_or(true)
})
.filter(|fs| {
filter_token
.map(|t| fs.get("CreationToken").and_then(Value::as_str) == Some(t))
.unwrap_or(true)
})
.map(|fs| with_live_mount_count(fs, data))
.collect();
let (page, next) =
paginate_marker(rows, query_one(q, "Marker"), query_one(q, "MaxItems"), 100);
let mut out = Map::new();
out.insert("FileSystems".into(), Value::Array(page));
if let Some(n) = next {
out.insert("NextMarker".into(), json!(n));
}
ok(StatusCode::OK, Value::Object(out))
}
fn require_fs<'a>(
&self,
data: &'a mut EfsData,
fsid: &str,
) -> Result<&'a mut Value, AwsServiceError> {
data.file_systems
.get_mut(fsid)
.ok_or_else(|| fs_not_found(fsid))
}
fn delete_file_system(&self, ctx: &Ctx, label: &str) -> Result<AwsResponse, AwsServiceError> {
let fsid = normalize_fs_id(label);
let mut guard = self.state.write();
let data = self.account(&mut guard, ctx);
if !data.file_systems.contains_key(&fsid) {
return Err(fs_not_found(&fsid));
}
let has_mt = data
.mount_targets
.values()
.any(|mt| mt.get("FileSystemId").and_then(Value::as_str) == Some(fsid.as_str()));
let has_ap = data.access_points.values().any(|ap| {
normalize_fs_id(ap.get("FileSystemId").and_then(Value::as_str).unwrap_or("")) == fsid
});
if has_mt || has_ap {
return Err(conflict(
"FileSystemInUse",
&format!("File system '{fsid}' is in use and cannot be deleted."),
));
}
data.file_systems.remove(&fsid);
data.tags.remove(&fsid);
data.lifecycle_configs.remove(&fsid);
data.backup_policies.remove(&fsid);
data.file_system_policies.remove(&fsid);
data.replications.remove(&fsid);
empty(StatusCode::NO_CONTENT)
}
fn update_file_system(
&self,
ctx: &Ctx,
label: &str,
b: &Value,
) -> Result<AwsResponse, AwsServiceError> {
let fsid = normalize_fs_id(label);
let mut guard = self.state.write();
let data = self.account(&mut guard, ctx);
let fs = self.require_fs(data, &fsid)?;
let obj = fs.as_object_mut().unwrap();
if obj.get("LifeCycleState").and_then(Value::as_str) == Some("creating") {
obj.insert("LifeCycleState".into(), json!("available"));
}
if let Some(tm) = b.get("ThroughputMode") {
obj.insert("ThroughputMode".into(), tm.clone());
}
if let Some(p) = b.get("ProvisionedThroughputInMibps") {
obj.insert("ProvisionedThroughputInMibps".into(), p.clone());
}
if obj.get("ThroughputMode").and_then(Value::as_str) != Some("provisioned") {
obj.remove("ProvisionedThroughputInMibps");
}
let fs_snapshot = fs.clone();
let updated = with_live_mount_count(&fs_snapshot, data);
ok(StatusCode::ACCEPTED, updated)
}
fn update_file_system_protection(
&self,
ctx: &Ctx,
label: &str,
b: &Value,
) -> Result<AwsResponse, AwsServiceError> {
let fsid = normalize_fs_id(label);
let mut guard = self.state.write();
let data = self.account(&mut guard, ctx);
let fs = self.require_fs(data, &fsid)?;
let protection = b
.get("ReplicationOverwriteProtection")
.and_then(Value::as_str)
.unwrap_or("ENABLED");
fs.as_object_mut().unwrap().insert(
"FileSystemProtection".into(),
json!({ "ReplicationOverwriteProtection": protection }),
);
ok(
StatusCode::OK,
json!({ "ReplicationOverwriteProtection": protection }),
)
}
}
fn tags_list_from_map(m: &Map<String, Value>) -> Value {
Value::Array(
m.iter()
.map(|(k, v)| json!({ "Key": k, "Value": v }))
.collect(),
)
}
fn with_live_mount_count(fs: &Value, data: &EfsData) -> Value {
let mut obj = fs.as_object().cloned().unwrap_or_default();
let fsid = obj
.get("FileSystemId")
.and_then(Value::as_str)
.unwrap_or("");
let count = data
.mount_targets
.values()
.filter(|mt| mt.get("FileSystemId").and_then(Value::as_str) == Some(fsid))
.count();
obj.insert("NumberOfMountTargets".into(), json!(count));
Value::Object(obj)
}
fn paginate_marker(
rows: Vec<Value>,
marker: Option<&str>,
max: Option<&str>,
default_max: usize,
) -> (Vec<Value>, Option<String>) {
let start = marker.and_then(|m| m.parse::<usize>().ok()).unwrap_or(0);
let max = max
.and_then(|m| m.parse::<usize>().ok())
.map(|m| m.max(1))
.unwrap_or(default_max);
let end = start.saturating_add(max).min(rows.len());
let page = rows.get(start..end).unwrap_or(&[]).to_vec();
let next = if end < rows.len() {
Some(end.to_string())
} else {
None
};
(page, next)
}
impl EfsService {
fn create_mount_target(&self, ctx: &Ctx, b: &Value) -> Result<AwsResponse, AwsServiceError> {
let fsid = normalize_fs_id(b.get("FileSystemId").and_then(Value::as_str).unwrap_or(""));
let subnet_id = b
.get("SubnetId")
.and_then(Value::as_str)
.unwrap_or("")
.to_string();
let h = hash_str(&subnet_id);
let (az_name, az_id, vpc_id) = match self.resolve_subnet(&ctx.account, &subnet_id) {
Some((az, azid, vpc)) => (az, azid, vpc),
None => {
if self.ec2_state.is_some() {
return Err(subnet_not_found(&subnet_id));
}
let az_index = (h % 3) as u8;
(
format!("{}{}", ctx.region, (b'a' + az_index) as char),
format!("{}-az{}", ctx.region, az_index + 1),
format!("vpc-{:017x}", h & 0x000f_ffff_ffff_ffff),
)
}
};
let mut guard = self.state.write();
let data = self.account(&mut guard, ctx);
let fs = data
.file_systems
.get(&fsid)
.ok_or_else(|| fs_not_found(&fsid))?;
let fs_state = fs
.get("LifeCycleState")
.and_then(Value::as_str)
.unwrap_or("");
if fs_state != "available" {
return Err(incorrect_fs_lifecycle_state(&fsid, fs_state));
}
for mt in data.mount_targets.values() {
if mt.get("FileSystemId").and_then(Value::as_str) == Some(fsid.as_str())
&& mt.get("AvailabilityZoneName").and_then(Value::as_str) == Some(az_name.as_str())
{
return Err(conflict(
"MountTargetConflict",
"A mount target already exists in this Availability Zone for the file system.",
));
}
}
let mtid = format!("fsmt-{}", hex17());
let ip = b
.get("IpAddress")
.and_then(Value::as_str)
.map(str::to_string)
.unwrap_or_else(|| format!("10.0.{}.{}", (h >> 8) % 256, h % 254 + 1));
let eni_id = format!("eni-{}", hex17());
let security_groups: Vec<String> = b
.get("SecurityGroups")
.and_then(Value::as_array)
.map(|a| {
a.iter()
.filter_map(|v| v.as_str().map(str::to_string))
.collect()
})
.filter(|v: &Vec<String>| !v.is_empty())
.unwrap_or_else(|| vec![format!("sg-{:017x}", h & 0x000f_ffff_ffff_ffff)]);
let mut mt = Map::new();
mt.insert("OwnerId".into(), json!(ctx.account));
mt.insert("MountTargetId".into(), json!(mtid));
mt.insert("FileSystemId".into(), json!(fsid));
mt.insert("SubnetId".into(), json!(subnet_id));
mt.insert("LifeCycleState".into(), json!("creating"));
mt.insert("IpAddress".into(), json!(ip));
mt.insert("NetworkInterfaceId".into(), json!(eni_id));
mt.insert("AvailabilityZoneId".into(), json!(az_id));
mt.insert("AvailabilityZoneName".into(), json!(az_name));
mt.insert("VpcId".into(), json!(vpc_id));
if let Some(ip6) = b.get("Ipv6Address") {
mt.insert("Ipv6Address".into(), ip6.clone());
}
data.mount_target_security_groups
.insert(mtid.clone(), security_groups);
data.mount_targets
.insert(mtid.clone(), Value::Object(mt.clone()));
ok(StatusCode::OK, Value::Object(mt))
}
fn describe_mount_targets(
&self,
ctx: &Ctx,
q: &[(String, String)],
) -> Result<AwsResponse, AwsServiceError> {
let by_fs = query_one(q, "FileSystemId").map(normalize_fs_id);
let by_mt = query_one(q, "MountTargetId");
let by_ap = query_one(q, "AccessPointId");
let filter_count = usize::from(by_fs.is_some())
+ usize::from(by_mt.is_some())
+ usize::from(by_ap.is_some());
if filter_count != 1 {
return Err(bad_request(
"Exactly one of FileSystemId, MountTargetId, or AccessPointId must be specified.",
));
}
let mut guard = self.state.write();
let data = self.account(&mut guard, ctx);
data.reconcile_lifecycle();
let ap_fs = if let Some(ap) = by_ap {
match data.access_points.get(ap) {
Some(ap_obj) => Some(normalize_fs_id(
ap_obj
.get("FileSystemId")
.and_then(Value::as_str)
.unwrap_or(""),
)),
None => return Err(ap_not_found(ap)),
}
} else {
None
};
if let Some(fsid) = &by_fs {
if !data.file_systems.contains_key(fsid) {
return Err(fs_not_found(fsid));
}
}
if let Some(mtid) = by_mt {
if !data.mount_targets.contains_key(mtid) {
return Err(mt_not_found(mtid));
}
}
let rows: Vec<Value> = data
.mount_targets
.iter()
.filter(|(mtid, mt)| {
let fs = mt.get("FileSystemId").and_then(Value::as_str);
by_fs.as_deref().map(|f| fs == Some(f)).unwrap_or(true)
&& by_mt.map(|m| mtid.as_str() == m).unwrap_or(true)
&& ap_fs.as_deref().map(|f| fs == Some(f)).unwrap_or(true)
})
.map(|(_, mt)| mt.clone())
.collect();
let (page, next) =
paginate_marker(rows, query_one(q, "Marker"), query_one(q, "MaxItems"), 10);
let mut out = Map::new();
out.insert("MountTargets".into(), Value::Array(page));
if let Some(n) = next {
out.insert("NextMarker".into(), json!(n));
}
ok(StatusCode::OK, Value::Object(out))
}
fn delete_mount_target(&self, ctx: &Ctx, label: &str) -> Result<AwsResponse, AwsServiceError> {
let mut guard = self.state.write();
let data = self.account(&mut guard, ctx);
if data.mount_targets.remove(label).is_none() {
return Err(mt_not_found(label));
}
data.mount_target_security_groups.remove(label);
empty(StatusCode::NO_CONTENT)
}
fn describe_mt_security_groups(
&self,
ctx: &Ctx,
label: &str,
) -> Result<AwsResponse, AwsServiceError> {
let mut guard = self.state.write();
let data = self.account(&mut guard, ctx);
data.reconcile_lifecycle();
if !data.mount_targets.contains_key(label) {
return Err(mt_not_found(label));
}
let sgs = data
.mount_target_security_groups
.get(label)
.cloned()
.unwrap_or_default();
ok(StatusCode::OK, json!({ "SecurityGroups": sgs }))
}
fn modify_mt_security_groups(
&self,
ctx: &Ctx,
label: &str,
b: &Value,
) -> Result<AwsResponse, AwsServiceError> {
let mut guard = self.state.write();
let data = self.account(&mut guard, ctx);
if !data.mount_targets.contains_key(label) {
return Err(mt_not_found(label));
}
let sgs: Vec<String> = b
.get("SecurityGroups")
.and_then(Value::as_array)
.map(|a| {
a.iter()
.filter_map(|v| v.as_str().map(str::to_string))
.collect()
})
.unwrap_or_default();
data.mount_target_security_groups
.insert(label.to_string(), sgs);
empty(StatusCode::NO_CONTENT)
}
}
impl EfsService {
fn create_access_point(&self, ctx: &Ctx, b: &Value) -> Result<AwsResponse, AwsServiceError> {
let client_token = b
.get("ClientToken")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string();
let fsid = normalize_fs_id(b.get("FileSystemId").and_then(Value::as_str).unwrap_or(""));
let mut guard = self.state.write();
let data = self.account(&mut guard, ctx);
if !data.file_systems.contains_key(&fsid) {
return Err(fs_not_found(&fsid));
}
if let Some((existing_id, _)) = data.access_points.iter().find(|(_, ap)| {
ap.get("ClientToken").and_then(Value::as_str) == Some(client_token.as_str())
}) {
let existing_id = existing_id.clone();
return Err(AwsServiceError::aws_error_with_fields(
StatusCode::CONFLICT,
"AccessPointAlreadyExists",
format!("Access point already exists with client token {client_token}"),
vec![("AccessPointId".to_string(), existing_id)],
));
}
let apid = format!("fsap-{}", hex17());
let tag_map = tags_to_map(b);
let mut ap = Map::new();
ap.insert("ClientToken".into(), json!(client_token));
if let Some(name) = tag_map.get("Name").and_then(Value::as_str) {
ap.insert("Name".into(), json!(name));
}
ap.insert("Tags".into(), tags_list_from_map(&tag_map));
ap.insert("AccessPointId".into(), json!(apid));
ap.insert("AccessPointArn".into(), json!(ap_arn(ctx, &apid)));
ap.insert("FileSystemId".into(), json!(fsid));
if let Some(pu) = b.get("PosixUser") {
ap.insert("PosixUser".into(), pu.clone());
}
if let Some(rd) = b.get("RootDirectory") {
ap.insert("RootDirectory".into(), rd.clone());
} else {
ap.insert("RootDirectory".into(), json!({ "Path": "/" }));
}
ap.insert("OwnerId".into(), json!(ctx.account));
ap.insert("LifeCycleState".into(), json!("creating"));
if !tag_map.is_empty() {
let entry = data.tags.entry(apid.clone()).or_default();
for (k, v) in &tag_map {
if let Some(vs) = v.as_str() {
entry.insert(k.clone(), vs.to_string());
}
}
}
data.access_points
.insert(apid.clone(), Value::Object(ap.clone()));
ok(StatusCode::OK, Value::Object(ap))
}
fn describe_access_points(
&self,
ctx: &Ctx,
q: &[(String, String)],
) -> Result<AwsResponse, AwsServiceError> {
let by_ap = query_one(q, "AccessPointId");
let by_fs = query_one(q, "FileSystemId").map(normalize_fs_id);
let mut guard = self.state.write();
let data = self.account(&mut guard, ctx);
data.reconcile_lifecycle();
if let Some(ap) = by_ap {
if !data.access_points.contains_key(ap) {
return Err(ap_not_found(ap));
}
}
if let Some(fsid) = &by_fs {
if !data.file_systems.contains_key(fsid) {
return Err(fs_not_found(fsid));
}
}
let rows: Vec<Value> = data
.access_points
.iter()
.filter(|(apid, ap)| {
by_ap.map(|a| apid.as_str() == a).unwrap_or(true)
&& by_fs
.as_deref()
.map(|f| ap.get("FileSystemId").and_then(Value::as_str) == Some(f))
.unwrap_or(true)
})
.map(|(_, ap)| ap.clone())
.collect();
let start = query_one(q, "NextToken")
.and_then(|t| t.parse::<usize>().ok())
.unwrap_or(0);
let max = query_one(q, "MaxResults")
.and_then(|m| m.parse::<usize>().ok())
.map(|m| m.max(1))
.unwrap_or(100);
let end = start.saturating_add(max).min(rows.len());
let page = rows.get(start..end).unwrap_or(&[]).to_vec();
let mut out = Map::new();
out.insert("AccessPoints".into(), Value::Array(page));
if end < rows.len() {
out.insert("NextToken".into(), json!(end.to_string()));
}
ok(StatusCode::OK, Value::Object(out))
}
fn delete_access_point(&self, ctx: &Ctx, label: &str) -> Result<AwsResponse, AwsServiceError> {
let mut guard = self.state.write();
let data = self.account(&mut guard, ctx);
if data.access_points.remove(label).is_none() {
return Err(ap_not_found(label));
}
data.tags.remove(label);
empty(StatusCode::NO_CONTENT)
}
}
impl EfsService {
fn put_lifecycle_configuration(
&self,
ctx: &Ctx,
label: &str,
b: &Value,
) -> Result<AwsResponse, AwsServiceError> {
let fsid = normalize_fs_id(label);
let mut guard = self.state.write();
let data = self.account(&mut guard, ctx);
if !data.file_systems.contains_key(&fsid) {
return Err(fs_not_found(&fsid));
}
let policies = b.get("LifecyclePolicies").cloned().unwrap_or(json!([]));
data.lifecycle_configs.insert(fsid, policies.clone());
ok(StatusCode::OK, json!({ "LifecyclePolicies": policies }))
}
fn describe_lifecycle_configuration(
&self,
ctx: &Ctx,
label: &str,
) -> Result<AwsResponse, AwsServiceError> {
let fsid = normalize_fs_id(label);
let guard = self.state.read();
let data = guard.get(&ctx.account);
let exists = data
.map(|d| d.file_systems.contains_key(&fsid))
.unwrap_or(false);
if !exists {
return Err(fs_not_found(&fsid));
}
let policies = data
.and_then(|d| d.lifecycle_configs.get(&fsid).cloned())
.unwrap_or(json!([]));
ok(StatusCode::OK, json!({ "LifecyclePolicies": policies }))
}
fn put_backup_policy(
&self,
ctx: &Ctx,
label: &str,
b: &Value,
) -> Result<AwsResponse, AwsServiceError> {
let fsid = normalize_fs_id(label);
let mut guard = self.state.write();
let data = self.account(&mut guard, ctx);
if !data.file_systems.contains_key(&fsid) {
return Err(fs_not_found(&fsid));
}
let status = b
.get("BackupPolicy")
.and_then(|p| p.get("Status"))
.and_then(Value::as_str)
.unwrap_or("DISABLED")
.to_string();
data.backup_policies.insert(fsid, status.clone());
ok(
StatusCode::OK,
json!({ "BackupPolicy": { "Status": status } }),
)
}
fn describe_backup_policy(
&self,
ctx: &Ctx,
label: &str,
) -> Result<AwsResponse, AwsServiceError> {
let fsid = normalize_fs_id(label);
let guard = self.state.read();
let data = guard.get(&ctx.account);
let exists = data
.map(|d| d.file_systems.contains_key(&fsid))
.unwrap_or(false);
if !exists {
return Err(fs_not_found(&fsid));
}
let status = data
.and_then(|d| d.backup_policies.get(&fsid).cloned())
.unwrap_or_else(|| "DISABLED".to_string());
ok(
StatusCode::OK,
json!({ "BackupPolicy": { "Status": status } }),
)
}
fn put_file_system_policy(
&self,
ctx: &Ctx,
label: &str,
b: &Value,
) -> Result<AwsResponse, AwsServiceError> {
let fsid = normalize_fs_id(label);
let mut guard = self.state.write();
let data = self.account(&mut guard, ctx);
if !data.file_systems.contains_key(&fsid) {
return Err(fs_not_found(&fsid));
}
let policy = b
.get("Policy")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string();
data.file_system_policies
.insert(fsid.clone(), policy.clone());
ok(
StatusCode::OK,
json!({ "FileSystemId": fsid, "Policy": policy }),
)
}
fn describe_file_system_policy(
&self,
ctx: &Ctx,
label: &str,
) -> Result<AwsResponse, AwsServiceError> {
let fsid = normalize_fs_id(label);
let guard = self.state.read();
let data = guard.get(&ctx.account);
let exists = data
.map(|d| d.file_systems.contains_key(&fsid))
.unwrap_or(false);
if !exists {
return Err(fs_not_found(&fsid));
}
match data.and_then(|d| d.file_system_policies.get(&fsid).cloned()) {
Some(policy) => ok(
StatusCode::OK,
json!({ "FileSystemId": fsid, "Policy": policy }),
),
None => Err(AwsServiceError::aws_error(
StatusCode::NOT_FOUND,
"PolicyNotFound",
format!("No policy is attached to file system '{fsid}'."),
)),
}
}
fn delete_file_system_policy(
&self,
ctx: &Ctx,
label: &str,
) -> Result<AwsResponse, AwsServiceError> {
let fsid = normalize_fs_id(label);
let mut guard = self.state.write();
let data = self.account(&mut guard, ctx);
if !data.file_systems.contains_key(&fsid) {
return Err(fs_not_found(&fsid));
}
data.file_system_policies.remove(&fsid);
empty(StatusCode::OK)
}
}
impl EfsService {
fn create_replication_configuration(
&self,
ctx: &Ctx,
label: &str,
b: &Value,
) -> Result<AwsResponse, AwsServiceError> {
let fsid = normalize_fs_id(label);
let mut guard = self.state.write();
let data = self.account(&mut guard, ctx);
if !data.file_systems.contains_key(&fsid) {
return Err(fs_not_found(&fsid));
}
let destinations: Vec<Value> = b
.get("Destinations")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default()
.iter()
.map(|d| {
let region = d
.get("Region")
.and_then(Value::as_str)
.unwrap_or(&ctx.region)
.to_string();
let dest_fs = d
.get("FileSystemId")
.and_then(Value::as_str)
.map(normalize_fs_id)
.unwrap_or_else(|| format!("fs-{}", hex17()));
let mut dest = Map::new();
dest.insert("Status".into(), json!("ENABLED"));
dest.insert("FileSystemId".into(), json!(dest_fs));
dest.insert("Region".into(), json!(region));
dest.insert("LastReplicatedTimestamp".into(), json!(now_ts()));
dest.insert("OwnerId".into(), json!(ctx.account));
if let Some(role) = d.get("RoleArn") {
dest.insert("RoleArn".into(), role.clone());
}
Value::Object(dest)
})
.collect();
let desc = json!({
"SourceFileSystemId": fsid,
"SourceFileSystemRegion": ctx.region,
"SourceFileSystemArn": fs_arn(ctx, &fsid),
"OriginalSourceFileSystemArn": fs_arn(ctx, &fsid),
"CreationTime": now_ts(),
"Destinations": destinations,
"SourceFileSystemOwnerId": ctx.account,
});
data.replications.insert(fsid, desc.clone());
ok(StatusCode::OK, desc)
}
fn describe_replication_configurations(
&self,
ctx: &Ctx,
q: &[(String, String)],
) -> Result<AwsResponse, AwsServiceError> {
let by_fs = query_one(q, "FileSystemId").map(normalize_fs_id);
let guard = self.state.read();
let data = guard.get(&ctx.account);
if let Some(fsid) = &by_fs {
let fs_exists = data
.map(|d| d.file_systems.contains_key(fsid))
.unwrap_or(false);
let has_replication = data
.map(|d| d.replications.contains_key(fsid))
.unwrap_or(false);
if !fs_exists && !has_replication {
return Err(fs_not_found(fsid));
}
if !has_replication {
return Err(replication_not_found(fsid));
}
}
let rows: Vec<Value> = data
.map(|d| {
d.replications
.iter()
.filter(|(src, _)| by_fs.as_deref().map(|f| src.as_str() == f).unwrap_or(true))
.map(|(_, r)| r.clone())
.collect()
})
.unwrap_or_default();
ok(StatusCode::OK, json!({ "Replications": rows }))
}
fn delete_replication_configuration(
&self,
ctx: &Ctx,
label: &str,
) -> Result<AwsResponse, AwsServiceError> {
let fsid = normalize_fs_id(label);
let mut guard = self.state.write();
let data = self.account(&mut guard, ctx);
if !data.file_systems.contains_key(&fsid) {
return Err(fs_not_found(&fsid));
}
if data.replications.remove(&fsid).is_none() {
return Err(AwsServiceError::aws_error(
StatusCode::NOT_FOUND,
"ReplicationNotFound",
format!("No replication configuration found for file system '{fsid}'."),
));
}
empty(StatusCode::NO_CONTENT)
}
}
impl EfsService {
fn create_tags(
&self,
ctx: &Ctx,
label: &str,
b: &Value,
) -> Result<AwsResponse, AwsServiceError> {
let fsid = normalize_fs_id(label);
let mut guard = self.state.write();
let data = self.account(&mut guard, ctx);
if !data.file_systems.contains_key(&fsid) {
return Err(fs_not_found(&fsid));
}
merge_tags(data, &fsid, b);
sync_resource_tags(data, &fsid);
empty(StatusCode::NO_CONTENT)
}
fn delete_tags(
&self,
ctx: &Ctx,
label: &str,
b: &Value,
) -> Result<AwsResponse, AwsServiceError> {
let fsid = normalize_fs_id(label);
let mut guard = self.state.write();
let data = self.account(&mut guard, ctx);
if !data.file_systems.contains_key(&fsid) {
return Err(fs_not_found(&fsid));
}
if let Some(keys) = b.get("TagKeys").and_then(Value::as_array) {
if let Some(entry) = data.tags.get_mut(&fsid) {
for k in keys {
if let Some(ks) = k.as_str() {
entry.remove(ks);
}
}
}
}
sync_resource_tags(data, &fsid);
empty(StatusCode::NO_CONTENT)
}
fn describe_tags(
&self,
ctx: &Ctx,
label: &str,
q: &[(String, String)],
) -> Result<AwsResponse, AwsServiceError> {
let fsid = normalize_fs_id(label);
let guard = self.state.read();
let data = guard.get(&ctx.account);
let exists = data
.map(|d| d.file_systems.contains_key(&fsid))
.unwrap_or(false);
if !exists {
return Err(fs_not_found(&fsid));
}
let all = data
.map(|d| tags_list(d, &fsid))
.unwrap_or_else(|| json!([]));
let rows = all.as_array().cloned().unwrap_or_default();
let (page, next) =
paginate_marker(rows, query_one(q, "Marker"), query_one(q, "MaxItems"), 100);
let mut out = Map::new();
out.insert("Tags".into(), Value::Array(page));
if let Some(n) = next {
out.insert("NextMarker".into(), json!(n));
}
ok(StatusCode::OK, Value::Object(out))
}
fn tag_resource(
&self,
ctx: &Ctx,
label: &str,
b: &Value,
) -> Result<AwsResponse, AwsServiceError> {
let rid = normalize_fs_id(label);
let mut guard = self.state.write();
let data = self.account(&mut guard, ctx);
self.require_taggable(data, &rid)?;
merge_tags(data, &rid, b);
sync_resource_tags(data, &rid);
empty(StatusCode::OK)
}
fn untag_resource(
&self,
ctx: &Ctx,
label: &str,
q: &[(String, String)],
) -> Result<AwsResponse, AwsServiceError> {
let rid = normalize_fs_id(label);
let mut guard = self.state.write();
let data = self.account(&mut guard, ctx);
self.require_taggable(data, &rid)?;
let keys = query_all(q, "tagKeys");
if let Some(entry) = data.tags.get_mut(&rid) {
for k in &keys {
entry.remove(k);
}
}
sync_resource_tags(data, &rid);
empty(StatusCode::OK)
}
fn list_tags_for_resource(
&self,
ctx: &Ctx,
label: &str,
q: &[(String, String)],
) -> Result<AwsResponse, AwsServiceError> {
let rid = normalize_fs_id(label);
let guard = self.state.read();
let data = guard.get(&ctx.account);
let known = data
.map(|d| {
(rid.starts_with("fsap-") && d.access_points.contains_key(&rid))
|| (rid.starts_with("fs-") && d.file_systems.contains_key(&rid))
})
.unwrap_or(false);
if !known {
return Err(if rid.starts_with("fsap-") {
ap_not_found(&rid)
} else {
fs_not_found(&rid)
});
}
let all = data
.map(|d| tags_list(d, &rid))
.unwrap_or_else(|| json!([]));
let rows = all.as_array().cloned().unwrap_or_default();
let start = query_one(q, "NextToken")
.and_then(|t| t.parse::<usize>().ok())
.unwrap_or(0);
let max = query_one(q, "MaxResults")
.and_then(|m| m.parse::<usize>().ok())
.map(|m| m.max(1))
.unwrap_or(100);
let end = start.saturating_add(max).min(rows.len());
let page = rows.get(start..end).unwrap_or(&[]).to_vec();
let mut out = Map::new();
out.insert("Tags".into(), Value::Array(page));
if end < rows.len() {
out.insert("NextToken".into(), json!(end.to_string()));
}
ok(StatusCode::OK, Value::Object(out))
}
fn require_taggable(&self, data: &EfsData, rid: &str) -> Result<(), AwsServiceError> {
if rid.starts_with("fsap-") {
if data.access_points.contains_key(rid) {
Ok(())
} else {
Err(ap_not_found(rid))
}
} else if data.file_systems.contains_key(rid) {
Ok(())
} else {
Err(fs_not_found(rid))
}
}
fn describe_account_preferences(&self, ctx: &Ctx) -> Result<AwsResponse, AwsServiceError> {
let guard = self.state.read();
match guard
.get(&ctx.account)
.and_then(|d| d.resource_id_preference.clone())
{
Some(pref) => ok(
StatusCode::OK,
json!({
"ResourceIdPreference": {
"ResourceIdType": pref,
"Resources": ["FILE_SYSTEM", "MOUNT_TARGET"]
}
}),
),
None => empty(StatusCode::OK),
}
}
fn put_account_preferences(
&self,
ctx: &Ctx,
b: &Value,
) -> Result<AwsResponse, AwsServiceError> {
let id_type = b
.get("ResourceIdType")
.and_then(Value::as_str)
.unwrap_or("LONG_ID")
.to_string();
let mut guard = self.state.write();
let data = self.account(&mut guard, ctx);
data.resource_id_preference = Some(id_type.clone());
ok(
StatusCode::OK,
json!({
"ResourceIdPreference": {
"ResourceIdType": id_type,
"Resources": ["FILE_SYSTEM", "MOUNT_TARGET"]
}
}),
)
}
}
fn merge_tags(data: &mut EfsData, resource_id: &str, b: &Value) {
let map = tags_to_map(b);
if map.is_empty() {
return;
}
let entry = data.tags.entry(resource_id.to_string()).or_default();
for (k, v) in &map {
if let Some(vs) = v.as_str() {
entry.insert(k.clone(), vs.to_string());
}
}
}
fn sync_resource_tags(data: &mut EfsData, rid: &str) {
let tags = data.tags.get(rid).cloned().unwrap_or_default();
let obj = if rid.starts_with("fsap-") {
data.access_points.get_mut(rid)
} else {
data.file_systems.get_mut(rid)
}
.and_then(Value::as_object_mut);
if let Some(obj) = obj {
match tags.get("Name") {
Some(name) => {
obj.insert("Name".into(), json!(name));
}
None => {
obj.remove("Name");
}
}
obj.insert(
"Tags".into(),
Value::Array(
tags.iter()
.map(|(k, v)| json!({ "Key": k, "Value": v }))
.collect(),
),
);
}
}
#[cfg(test)]
mod tests {
use super::*;
use parking_lot::RwLock;
fn ctx() -> Ctx {
Ctx {
account: "000000000000".to_string(),
region: "us-east-1".to_string(),
}
}
fn svc() -> EfsService {
let state: SharedEfsState = Arc::new(RwLock::new(MultiAccountState::new(
"000000000000",
"us-east-1",
"",
)));
EfsService::new(state)
}
fn empty_ec2() -> fakecloud_ec2::SharedEc2State {
Arc::new(RwLock::new(MultiAccountState::new(
"000000000000",
"us-east-1",
"",
)))
}
fn seed_fs(s: &EfsService, fsid: &str, life_cycle: &str) {
let mut g = s.state.write();
let d = g.get_or_create("000000000000");
d.file_systems.insert(
fsid.to_string(),
json!({ "FileSystemId": fsid, "LifeCycleState": life_cycle }),
);
}
fn body_value(resp: &AwsResponse) -> Value {
serde_json::from_slice(resp.body.expect_bytes()).unwrap()
}
#[test]
fn create_mount_target_nonexistent_subnet_is_subnet_not_found() {
let s = svc().with_ec2_state(empty_ec2());
seed_fs(&s, "fs-1", "available");
let err = s
.create_mount_target(
&ctx(),
&json!({ "FileSystemId": "fs-1", "SubnetId": "subnet-does-not-exist" }),
)
.err()
.unwrap();
assert_eq!(err.code(), "SubnetNotFound");
assert_eq!(err.status(), StatusCode::BAD_REQUEST);
}
#[test]
fn create_mount_target_without_ec2_synthesizes_subnet() {
let s = svc();
seed_fs(&s, "fs-1", "available");
let resp = s
.create_mount_target(
&ctx(),
&json!({ "FileSystemId": "fs-1", "SubnetId": "subnet-synth" }),
)
.unwrap();
assert!(resp.status.is_success());
}
#[test]
fn create_mount_target_on_creating_fs_is_incorrect_lifecycle() {
let s = svc();
seed_fs(&s, "fs-1", "creating");
let err = s
.create_mount_target(
&ctx(),
&json!({ "FileSystemId": "fs-1", "SubnetId": "subnet-1" }),
)
.err()
.unwrap();
assert_eq!(err.code(), "IncorrectFileSystemLifeCycleState");
assert_eq!(err.status(), StatusCode::CONFLICT);
}
#[test]
fn delete_file_system_with_access_point_is_in_use() {
let s = svc();
seed_fs(&s, "fs-1", "available");
{
let mut g = s.state.write();
let d = g.get_or_create("000000000000");
d.access_points.insert(
"fsap-1".to_string(),
json!({ "AccessPointId": "fsap-1", "FileSystemId": "fs-1" }),
);
}
let err = s.delete_file_system(&ctx(), "fs-1").err().unwrap();
assert_eq!(err.code(), "FileSystemInUse");
assert_eq!(err.status(), StatusCode::CONFLICT);
}
#[test]
fn describe_replication_missing_is_replication_not_found() {
let s = svc();
seed_fs(&s, "fs-1", "available");
let err = s
.describe_replication_configurations(
&ctx(),
&[("FileSystemId".to_string(), "fs-1".to_string())],
)
.err()
.unwrap();
assert_eq!(err.code(), "ReplicationNotFound");
assert_eq!(err.status(), StatusCode::NOT_FOUND);
}
#[test]
fn describe_replication_no_filter_is_empty_ok() {
let s = svc();
let resp = s.describe_replication_configurations(&ctx(), &[]).unwrap();
assert!(resp.status.is_success());
assert_eq!(
body_value(&resp)["Replications"].as_array().unwrap().len(),
0
);
}
#[test]
fn describe_mount_targets_two_filters_is_bad_request() {
let s = svc();
let err = s
.describe_mount_targets(
&ctx(),
&[
("FileSystemId".to_string(), "fs-1".to_string()),
("MountTargetId".to_string(), "fsmt-1".to_string()),
],
)
.err()
.unwrap();
assert_eq!(err.code(), "BadRequest");
assert_eq!(err.status(), StatusCode::BAD_REQUEST);
}
#[test]
fn create_file_system_provisioned_without_mibps_is_bad_request() {
let s = svc();
let err = s
.create_file_system(
&ctx(),
&json!({ "CreationToken": "tok-1", "ThroughputMode": "provisioned" }),
)
.err()
.unwrap();
assert_eq!(err.code(), "BadRequest");
let err2 = s
.create_file_system(
&ctx(),
&json!({
"CreationToken": "tok-2",
"ThroughputMode": "bursting",
"ProvisionedThroughputInMibps": 128
}),
)
.err()
.unwrap();
assert_eq!(err2.code(), "BadRequest");
let ok = s
.create_file_system(
&ctx(),
&json!({
"CreationToken": "tok-3",
"ThroughputMode": "provisioned",
"ProvisionedThroughputInMibps": 256
}),
)
.unwrap();
assert!(ok.status.is_success());
}
#[test]
fn describe_account_preferences_fresh_account_is_empty() {
let s = svc();
let resp = s.describe_account_preferences(&ctx()).unwrap();
assert!(resp.status.is_success());
let v = body_value(&resp);
assert!(
v.get("ResourceIdPreference").is_none(),
"expected empty {{}}, got {v}"
);
s.put_account_preferences(&ctx(), &json!({ "ResourceIdType": "SHORT_ID" }))
.unwrap();
let resp2 = s.describe_account_preferences(&ctx()).unwrap();
let v2 = body_value(&resp2);
assert_eq!(v2["ResourceIdPreference"]["ResourceIdType"], "SHORT_ID");
}
#[test]
fn access_point_tags_resync_agrees() {
let s = svc();
seed_fs(&s, "fs-1", "available");
let created = s
.create_access_point(
&ctx(),
&json!({ "ClientToken": "ct-1", "FileSystemId": "fs-1" }),
)
.unwrap();
let apid = body_value(&created)["AccessPointId"]
.as_str()
.unwrap()
.to_string();
s.tag_resource(
&ctx(),
&apid,
&json!({ "Tags": [{ "Key": "Env", "Value": "prod" }] }),
)
.unwrap();
let embedded: Vec<(String, String)> = {
let g = s.state.read();
let ap = g
.get("000000000000")
.unwrap()
.access_points
.get(&apid)
.unwrap()
.clone();
ap["Tags"]
.as_array()
.unwrap()
.iter()
.map(|t| {
(
t["Key"].as_str().unwrap().to_string(),
t["Value"].as_str().unwrap().to_string(),
)
})
.collect()
};
let listed_resp = s.list_tags_for_resource(&ctx(), &apid, &[]).unwrap();
let listed: Vec<(String, String)> = body_value(&listed_resp)["Tags"]
.as_array()
.unwrap()
.iter()
.map(|t| {
(
t["Key"].as_str().unwrap().to_string(),
t["Value"].as_str().unwrap().to_string(),
)
})
.collect();
assert_eq!(embedded, vec![("Env".to_string(), "prod".to_string())]);
assert_eq!(
embedded, listed,
"embedded Tags must match ListTagsForResource"
);
s.untag_resource(&ctx(), &apid, &[("tagKeys".to_string(), "Env".to_string())])
.unwrap();
let g = s.state.read();
let ap = g
.get("000000000000")
.unwrap()
.access_points
.get(&apid)
.unwrap();
assert_eq!(ap["Tags"].as_array().unwrap().len(), 0);
}
}