use chrono::Utc;
use uuid::Uuid;
use fakecloud_aws::arn::Arn;
use fakecloud_core::service::{AwsRequest, AwsResponse, AwsServiceError};
use super::helpers::*;
use super::RedshiftService;
use crate::state::{
HsmClientCertificate, HsmConfiguration, LoggingStatus, ScheduledAction, SnapshotCopyGrant,
SnapshotCopyStatus, SnapshotSchedule, UsageLimit,
};
fn render_hsm_client_certificate(c: &HsmClientCertificate) -> String {
format!(
"<HsmClientCertificateIdentifier>{id}</HsmClientCertificateIdentifier>\
<HsmClientCertificatePublicKey>{key}</HsmClientCertificatePublicKey>{tags}",
id = xml_escape(&c.hsm_client_certificate_identifier),
key = xml_escape(&c.hsm_client_certificate_public_key),
tags = render_tags(&c.tags),
)
}
fn render_hsm_configuration(c: &HsmConfiguration) -> String {
format!(
"<HsmConfigurationIdentifier>{id}</HsmConfigurationIdentifier><Description>{desc}</Description>\
<HsmIpAddress>{ip}</HsmIpAddress><HsmPartitionName>{part}</HsmPartitionName>{tags}",
id = xml_escape(&c.hsm_configuration_identifier),
desc = xml_escape(&c.description),
ip = xml_escape(&c.hsm_ip_address),
part = xml_escape(&c.hsm_partition_name),
tags = render_tags(&c.tags),
)
}
fn render_snapshot_copy_grant(g: &SnapshotCopyGrant) -> String {
format!(
"<SnapshotCopyGrantName>{name}</SnapshotCopyGrantName><KmsKeyId>{kms}</KmsKeyId>{tags}",
name = xml_escape(&g.snapshot_copy_grant_name),
kms = xml_escape(&g.kms_key_id),
tags = render_tags(&g.tags),
)
}
fn render_snapshot_schedule(s: &SnapshotSchedule) -> String {
let defs: String = s
.schedule_definitions
.iter()
.map(|d| tag_elem("ScheduleDefinition", d))
.collect();
let clusters: String = s
.associated_clusters
.iter()
.map(|c| {
format!(
"<ClusterAssociatedToSchedule><ClusterIdentifier>{}</ClusterIdentifier><ScheduleAssociationState>ACTIVE</ScheduleAssociationState></ClusterAssociatedToSchedule>",
xml_escape(c)
)
})
.collect();
format!(
"<ScheduleIdentifier>{id}</ScheduleIdentifier><ScheduleDescription>{desc}</ScheduleDescription>\
<ScheduleDefinitions>{defs}</ScheduleDefinitions>\
<AssociatedClusterCount>{count}</AssociatedClusterCount>\
<AssociatedClusters>{clusters}</AssociatedClusters>{tags}",
id = xml_escape(&s.schedule_identifier),
desc = xml_escape(&s.schedule_description),
count = s.associated_cluster_count,
tags = render_tags(&s.tags),
)
}
fn render_scheduled_action(a: &ScheduledAction) -> String {
let target = a
.target_action
.as_ref()
.map(|t| format!("<TargetAction>{t}</TargetAction>"))
.unwrap_or_default();
format!(
"<ScheduledActionName>{name}</ScheduledActionName>{target}<Schedule>{schedule}</Schedule>\
<IamRole>{iam}</IamRole><ScheduledActionDescription>{desc}</ScheduledActionDescription>\
<State>{state}</State>{start}{end}",
name = xml_escape(&a.scheduled_action_name),
schedule = xml_escape(&a.schedule),
iam = xml_escape(&a.iam_role),
desc = xml_escape(&a.scheduled_action_description),
state = xml_escape(&a.state),
start = a
.start_time
.map(|t| format!(
"<StartTime>{}</StartTime>",
t.format("%Y-%m-%dT%H:%M:%S%.3fZ")
))
.unwrap_or_default(),
end = a
.end_time
.map(|t| format!("<EndTime>{}</EndTime>", t.format("%Y-%m-%dT%H:%M:%S%.3fZ")))
.unwrap_or_default(),
)
}
fn render_usage_limit(u: &UsageLimit) -> String {
format!(
"<UsageLimitId>{id}</UsageLimitId><ClusterIdentifier>{cluster}</ClusterIdentifier>\
<FeatureType>{feature}</FeatureType><LimitType>{ltype}</LimitType><Amount>{amount}</Amount>\
<Period>{period}</Period><BreachAction>{action}</BreachAction>{tags}",
id = xml_escape(&u.usage_limit_id),
cluster = xml_escape(&u.cluster_identifier),
feature = xml_escape(&u.feature_type),
ltype = xml_escape(&u.limit_type),
amount = u.amount,
period = xml_escape(&u.period),
action = xml_escape(&u.breach_action),
tags = render_tags(&u.tags),
)
}
fn render_logging(l: &LoggingStatus) -> String {
let log_exports = if l.log_exports.is_empty() {
String::new()
} else {
let inner: String = l
.log_exports
.iter()
.map(|e| tag_elem("member", e))
.collect();
format!("<LogExports>{inner}</LogExports>")
};
format!(
"<LoggingEnabled>{enabled}</LoggingEnabled>{bucket}{prefix}{dest}{log_exports}{time}",
enabled = l.logging_enabled,
bucket = opt_elem("BucketName", l.bucket_name.as_deref()),
prefix = opt_elem("S3KeyPrefix", l.s3_key_prefix.as_deref()),
dest = opt_elem("LogDestinationType", l.log_destination_type.as_deref()),
time = l
.last_successful_delivery_time
.map(|t| format!(
"<LastSuccessfulDeliveryTime>{}</LastSuccessfulDeliveryTime>",
t.format("%Y-%m-%dT%H:%M:%S%.3fZ")
))
.unwrap_or_default(),
)
}
impl RedshiftService {
pub(super) fn create_hsm_client_certificate(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let id = param(req, "HsmClientCertificateIdentifier").unwrap_or_default();
let mut guard = self.state.write();
let acct = guard.account(&req.account_id);
if acct.hsm_client_certificates.contains_key(&id) {
return Err(hsm_client_certificate_already_exists(&id));
}
let cert = HsmClientCertificate {
hsm_client_certificate_identifier: id.clone(),
hsm_client_certificate_public_key: format!(
"-----BEGIN CERTIFICATE-----\nFAKECLOUD{}\n-----END CERTIFICATE-----",
Uuid::new_v4().simple()
),
tags: parse_tags(req),
};
acct.hsm_client_certificates.insert(id, cert.clone());
Ok(xml_resp(
"CreateHsmClientCertificate",
format!(
"<HsmClientCertificate>{}</HsmClientCertificate>",
render_hsm_client_certificate(&cert)
),
&req.request_id,
))
}
pub(super) fn describe_hsm_client_certificates(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let guard = self.state.read();
let all: Vec<HsmClientCertificate> = guard
.accounts
.get(&req.account_id)
.map(|a| a.hsm_client_certificates.values().cloned().collect())
.unwrap_or_default();
let (page, next) = paginate(&all, req);
let inner: String = page
.iter()
.map(|c| {
format!(
"<HsmClientCertificate>{}</HsmClientCertificate>",
render_hsm_client_certificate(c)
)
})
.collect();
Ok(xml_resp(
"DescribeHsmClientCertificates",
format!(
"{}<HsmClientCertificates>{inner}</HsmClientCertificates>",
render_marker(next)
),
&req.request_id,
))
}
pub(super) fn delete_hsm_client_certificate(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let id = param(req, "HsmClientCertificateIdentifier").unwrap_or_default();
let mut guard = self.state.write();
let acct = guard.account(&req.account_id);
if acct.hsm_client_certificates.remove(&id).is_none() {
return Err(hsm_client_certificate_not_found(&id));
}
Ok(xml_metadata_only(
"DeleteHsmClientCertificate",
&req.request_id,
))
}
pub(super) fn create_hsm_configuration(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let id = param(req, "HsmConfigurationIdentifier").unwrap_or_default();
let mut guard = self.state.write();
let acct = guard.account(&req.account_id);
if acct.hsm_configurations.contains_key(&id) {
return Err(hsm_configuration_already_exists(&id));
}
let cfg = HsmConfiguration {
hsm_configuration_identifier: id.clone(),
description: param(req, "Description").unwrap_or_default(),
hsm_ip_address: param(req, "HsmIpAddress").unwrap_or_default(),
hsm_partition_name: param(req, "HsmPartitionName").unwrap_or_default(),
tags: parse_tags(req),
};
acct.hsm_configurations.insert(id, cfg.clone());
Ok(xml_resp(
"CreateHsmConfiguration",
format!(
"<HsmConfiguration>{}</HsmConfiguration>",
render_hsm_configuration(&cfg)
),
&req.request_id,
))
}
pub(super) fn describe_hsm_configurations(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let guard = self.state.read();
let all: Vec<HsmConfiguration> = guard
.accounts
.get(&req.account_id)
.map(|a| a.hsm_configurations.values().cloned().collect())
.unwrap_or_default();
let (page, next) = paginate(&all, req);
let inner: String = page
.iter()
.map(|c| {
format!(
"<HsmConfiguration>{}</HsmConfiguration>",
render_hsm_configuration(c)
)
})
.collect();
Ok(xml_resp(
"DescribeHsmConfigurations",
format!(
"{}<HsmConfigurations>{inner}</HsmConfigurations>",
render_marker(next)
),
&req.request_id,
))
}
pub(super) fn delete_hsm_configuration(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let id = param(req, "HsmConfigurationIdentifier").unwrap_or_default();
let mut guard = self.state.write();
let acct = guard.account(&req.account_id);
if acct.hsm_configurations.remove(&id).is_none() {
return Err(hsm_configuration_not_found(&id));
}
Ok(xml_metadata_only("DeleteHsmConfiguration", &req.request_id))
}
pub(super) fn create_snapshot_copy_grant(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let name = param(req, "SnapshotCopyGrantName").unwrap_or_default();
let mut guard = self.state.write();
let acct = guard.account(&req.account_id);
if acct.snapshot_copy_grants.contains_key(&name) {
return Err(snapshot_copy_grant_already_exists(&name));
}
let g = SnapshotCopyGrant {
snapshot_copy_grant_name: name.clone(),
kms_key_id: param_or(req, "KmsKeyId", "default"),
tags: parse_tags(req),
};
acct.snapshot_copy_grants.insert(name, g.clone());
Ok(xml_resp(
"CreateSnapshotCopyGrant",
format!(
"<SnapshotCopyGrant>{}</SnapshotCopyGrant>",
render_snapshot_copy_grant(&g)
),
&req.request_id,
))
}
pub(super) fn describe_snapshot_copy_grants(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let guard = self.state.read();
let all: Vec<SnapshotCopyGrant> = guard
.accounts
.get(&req.account_id)
.map(|a| a.snapshot_copy_grants.values().cloned().collect())
.unwrap_or_default();
let (page, next) = paginate(&all, req);
let inner: String = page
.iter()
.map(|g| {
format!(
"<SnapshotCopyGrant>{}</SnapshotCopyGrant>",
render_snapshot_copy_grant(g)
)
})
.collect();
Ok(xml_resp(
"DescribeSnapshotCopyGrants",
format!(
"{}<SnapshotCopyGrants>{inner}</SnapshotCopyGrants>",
render_marker(next)
),
&req.request_id,
))
}
pub(super) fn delete_snapshot_copy_grant(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let name = param(req, "SnapshotCopyGrantName").unwrap_or_default();
let mut guard = self.state.write();
let acct = guard.account(&req.account_id);
if acct.snapshot_copy_grants.remove(&name).is_none() {
return Err(snapshot_copy_grant_not_found(&name));
}
Ok(xml_metadata_only(
"DeleteSnapshotCopyGrant",
&req.request_id,
))
}
pub(super) fn enable_snapshot_copy(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
self.cluster_snapshot_copy(req, "EnableSnapshotCopy")
}
pub(super) fn disable_snapshot_copy(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
self.cluster_snapshot_copy(req, "DisableSnapshotCopy")
}
pub(super) fn modify_snapshot_copy_retention_period(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
self.cluster_snapshot_copy(req, "ModifySnapshotCopyRetentionPeriod")
}
fn cluster_snapshot_copy(
&self,
req: &AwsRequest,
action: &str,
) -> Result<AwsResponse, AwsServiceError> {
let id = param(req, "ClusterIdentifier").unwrap_or_default();
let mut guard = self.state.write();
let acct = guard.account(&req.account_id);
let c = acct
.clusters
.get_mut(&id)
.ok_or_else(|| cluster_not_found(&id))?;
match action {
"EnableSnapshotCopy" => {
c.snapshot_copy = Some(SnapshotCopyStatus {
destination_region: param(req, "DestinationRegion").unwrap_or_default(),
retention_period: long_param(req, "RetentionPeriod").unwrap_or(7),
manual_snapshot_retention_period: int_param(
req,
"ManualSnapshotRetentionPeriod",
)
.unwrap_or(-1),
snapshot_copy_grant_name: param(req, "SnapshotCopyGrantName"),
});
}
"DisableSnapshotCopy" => {
c.snapshot_copy = None;
}
"ModifySnapshotCopyRetentionPeriod" => {
if let Some(sc) = c.snapshot_copy.as_mut() {
if let Some(r) = long_param(req, "RetentionPeriod") {
sc.retention_period = r;
}
if bool_param(req, "Manual").unwrap_or(false) {
sc.manual_snapshot_retention_period =
long_param(req, "RetentionPeriod").unwrap_or(-1) as i32;
}
}
}
_ => {}
}
let cluster = c.clone();
Ok(xml_resp(
action,
format!(
"<Cluster>{}</Cluster>",
super::clusters::render_cluster(&cluster)
),
&req.request_id,
))
}
pub(super) fn create_snapshot_schedule(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let id = param(req, "ScheduleIdentifier")
.unwrap_or_else(|| format!("schedule-{}", &Uuid::new_v4().simple().to_string()[..8]));
let mut guard = self.state.write();
let acct = guard.account(&req.account_id);
if acct.snapshot_schedules.contains_key(&id) {
return Err(snapshot_schedule_already_exists(&id));
}
let s = SnapshotSchedule {
schedule_identifier: id.clone(),
schedule_description: param(req, "ScheduleDescription").unwrap_or_default(),
schedule_definitions: member_list(req, "ScheduleDefinitions", "ScheduleDefinition"),
tags: parse_tags(req),
associated_cluster_count: 0,
associated_clusters: Vec::new(),
};
acct.snapshot_schedules.insert(id, s.clone());
Ok(xml_resp(
"CreateSnapshotSchedule",
render_snapshot_schedule(&s),
&req.request_id,
))
}
pub(super) fn describe_snapshot_schedules(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let schedule_filter = param(req, "ScheduleIdentifier");
let cluster_filter = param(req, "ClusterIdentifier");
let guard = self.state.read();
let all: Vec<SnapshotSchedule> = guard
.accounts
.get(&req.account_id)
.map(|a| {
a.snapshot_schedules
.values()
.filter(|s| {
schedule_filter
.as_ref()
.map(|f| &s.schedule_identifier == f)
.unwrap_or(true)
})
.filter(|s| {
cluster_filter
.as_ref()
.map(|c| s.associated_clusters.contains(c))
.unwrap_or(true)
})
.cloned()
.collect()
})
.unwrap_or_default();
let (page, next) = paginate(&all, req);
let inner: String = page
.iter()
.map(|s| {
format!(
"<SnapshotSchedule>{}</SnapshotSchedule>",
render_snapshot_schedule(s)
)
})
.collect();
Ok(xml_resp(
"DescribeSnapshotSchedules",
format!(
"{}<SnapshotSchedules>{inner}</SnapshotSchedules>",
render_marker(next)
),
&req.request_id,
))
}
pub(super) fn modify_snapshot_schedule(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let id = param(req, "ScheduleIdentifier").unwrap_or_default();
let defs = member_list(req, "ScheduleDefinitions", "ScheduleDefinition");
let mut guard = self.state.write();
let acct = guard.account(&req.account_id);
let s = acct
.snapshot_schedules
.get_mut(&id)
.ok_or_else(|| snapshot_schedule_not_found(&id))?;
if !defs.is_empty() {
s.schedule_definitions = defs;
}
Ok(xml_resp(
"ModifySnapshotSchedule",
render_snapshot_schedule(s),
&req.request_id,
))
}
pub(super) fn delete_snapshot_schedule(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let id = param(req, "ScheduleIdentifier").unwrap_or_default();
let mut guard = self.state.write();
let acct = guard.account(&req.account_id);
if acct.snapshot_schedules.remove(&id).is_none() {
return Err(snapshot_schedule_not_found(&id));
}
Ok(xml_metadata_only("DeleteSnapshotSchedule", &req.request_id))
}
pub(super) fn create_scheduled_action(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let name = param(req, "ScheduledActionName").unwrap_or_default();
let mut guard = self.state.write();
let acct = guard.account(&req.account_id);
if acct.scheduled_actions.contains_key(&name) {
return Err(scheduled_action_already_exists(&name));
}
let a = ScheduledAction {
scheduled_action_name: name.clone(),
target_action: extract_target_action(req),
schedule: param(req, "Schedule").unwrap_or_default(),
iam_role: param(req, "IamRole").unwrap_or_default(),
scheduled_action_description: param(req, "ScheduledActionDescription")
.unwrap_or_default(),
state: if bool_param(req, "Enable").unwrap_or(true) {
"ACTIVE".to_string()
} else {
"DISABLED".to_string()
},
start_time: None,
end_time: None,
enable: bool_param(req, "Enable").unwrap_or(true),
};
acct.scheduled_actions.insert(name, a.clone());
Ok(xml_resp(
"CreateScheduledAction",
render_scheduled_action(&a),
&req.request_id,
))
}
pub(super) fn describe_scheduled_actions(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let guard = self.state.read();
let all: Vec<ScheduledAction> = guard
.accounts
.get(&req.account_id)
.map(|a| a.scheduled_actions.values().cloned().collect())
.unwrap_or_default();
let (page, next) = paginate(&all, req);
let inner: String = page
.iter()
.map(|a| {
format!(
"<ScheduledAction>{}</ScheduledAction>",
render_scheduled_action(a)
)
})
.collect();
Ok(xml_resp(
"DescribeScheduledActions",
format!(
"{}<ScheduledActions>{inner}</ScheduledActions>",
render_marker(next)
),
&req.request_id,
))
}
pub(super) fn modify_scheduled_action(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let name = param(req, "ScheduledActionName").unwrap_or_default();
let mut guard = self.state.write();
let acct = guard.account(&req.account_id);
let a = acct
.scheduled_actions
.get_mut(&name)
.ok_or_else(|| scheduled_action_not_found(&name))?;
if let Some(s) = param(req, "Schedule") {
a.schedule = s;
}
if let Some(r) = param(req, "IamRole") {
a.iam_role = r;
}
if let Some(d) = param(req, "ScheduledActionDescription") {
a.scheduled_action_description = d;
}
if let Some(e) = bool_param(req, "Enable") {
a.enable = e;
a.state = if e {
"ACTIVE".to_string()
} else {
"DISABLED".to_string()
};
}
if let Some(t) = extract_target_action(req) {
a.target_action = Some(t);
}
Ok(xml_resp(
"ModifyScheduledAction",
render_scheduled_action(a),
&req.request_id,
))
}
pub(super) fn delete_scheduled_action(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let name = param(req, "ScheduledActionName").unwrap_or_default();
let mut guard = self.state.write();
let acct = guard.account(&req.account_id);
if acct.scheduled_actions.remove(&name).is_none() {
return Err(scheduled_action_not_found(&name));
}
Ok(xml_metadata_only("DeleteScheduledAction", &req.request_id))
}
pub(super) fn create_usage_limit(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let cluster = param(req, "ClusterIdentifier").unwrap_or_default();
let feature = param(req, "FeatureType").unwrap_or_default();
let limit_type = param(req, "LimitType").unwrap_or_default();
let id = format!("usagelimit-{}", &Uuid::new_v4().simple().to_string()[..12]);
let mut guard = self.state.write();
let acct = guard.account(&req.account_id);
if acct.usage_limits.values().any(|u| {
u.cluster_identifier == cluster
&& u.feature_type == feature
&& u.limit_type == limit_type
}) {
return Err(usage_limit_already_exists(&feature));
}
let u = UsageLimit {
usage_limit_id: id.clone(),
cluster_identifier: cluster,
feature_type: feature,
limit_type,
amount: long_param(req, "Amount").unwrap_or(0),
period: param_or(req, "Period", "monthly"),
breach_action: param_or(req, "BreachAction", "log"),
tags: parse_tags(req),
};
acct.usage_limits.insert(id, u.clone());
Ok(xml_resp(
"CreateUsageLimit",
render_usage_limit(&u),
&req.request_id,
))
}
pub(super) fn describe_usage_limits(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let cluster_filter = param(req, "ClusterIdentifier");
let id_filter = param(req, "UsageLimitId");
let guard = self.state.read();
let all: Vec<UsageLimit> = guard
.accounts
.get(&req.account_id)
.map(|a| {
a.usage_limits
.values()
.filter(|u| {
cluster_filter
.as_ref()
.map(|c| &u.cluster_identifier == c)
.unwrap_or(true)
&& id_filter
.as_ref()
.map(|i| &u.usage_limit_id == i)
.unwrap_or(true)
})
.cloned()
.collect()
})
.unwrap_or_default();
let (page, next) = paginate(&all, req);
let inner: String = page
.iter()
.map(|u| format!("<member>{}</member>", render_usage_limit(u)))
.collect();
Ok(xml_resp(
"DescribeUsageLimits",
format!("{}<UsageLimits>{inner}</UsageLimits>", render_marker(next)),
&req.request_id,
))
}
pub(super) fn modify_usage_limit(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let id = param(req, "UsageLimitId").unwrap_or_default();
let mut guard = self.state.write();
let acct = guard.account(&req.account_id);
let u = acct
.usage_limits
.get_mut(&id)
.ok_or_else(|| usage_limit_not_found(&id))?;
if let Some(a) = long_param(req, "Amount") {
u.amount = a;
}
if let Some(b) = param(req, "BreachAction") {
u.breach_action = b;
}
Ok(xml_resp(
"ModifyUsageLimit",
render_usage_limit(u),
&req.request_id,
))
}
pub(super) fn delete_usage_limit(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let id = param(req, "UsageLimitId").unwrap_or_default();
let mut guard = self.state.write();
let acct = guard.account(&req.account_id);
if acct.usage_limits.remove(&id).is_none() {
return Err(usage_limit_not_found(&id));
}
Ok(xml_metadata_only("DeleteUsageLimit", &req.request_id))
}
pub(super) fn enable_logging(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
let id = param(req, "ClusterIdentifier").unwrap_or_default();
let mut guard = self.state.write();
let acct = guard.account(&req.account_id);
if !acct.clusters.contains_key(&id) {
return Err(cluster_not_found(&id));
}
let status = LoggingStatus {
logging_enabled: true,
bucket_name: param(req, "BucketName"),
s3_key_prefix: param(req, "S3KeyPrefix"),
log_destination_type: param(req, "LogDestinationType"),
log_exports: member_list(req, "LogExports", "member"),
last_successful_delivery_time: Some(Utc::now()),
};
acct.logging.insert(id, status.clone());
Ok(xml_resp(
"EnableLogging",
render_logging(&status),
&req.request_id,
))
}
pub(super) fn disable_logging(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
let id = param(req, "ClusterIdentifier").unwrap_or_default();
let mut guard = self.state.write();
let acct = guard.account(&req.account_id);
if !acct.clusters.contains_key(&id) {
return Err(cluster_not_found(&id));
}
acct.logging.remove(&id);
let status = LoggingStatus {
logging_enabled: false,
bucket_name: None,
s3_key_prefix: None,
log_destination_type: None,
log_exports: Vec::new(),
last_successful_delivery_time: None,
};
Ok(xml_resp(
"DisableLogging",
render_logging(&status),
&req.request_id,
))
}
pub(super) fn describe_logging_status(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let id = param(req, "ClusterIdentifier").unwrap_or_default();
let guard = self.state.read();
let acct = guard
.accounts
.get(&req.account_id)
.ok_or_else(|| cluster_not_found(&id))?;
if !acct.clusters.contains_key(&id) {
return Err(cluster_not_found(&id));
}
let status = acct.logging.get(&id).cloned().unwrap_or(LoggingStatus {
logging_enabled: false,
bucket_name: None,
s3_key_prefix: None,
log_destination_type: None,
log_exports: Vec::new(),
last_successful_delivery_time: None,
});
Ok(xml_resp(
"DescribeLoggingStatus",
render_logging(&status),
&req.request_id,
))
}
pub(super) fn put_resource_policy(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let arn = param(req, "ResourceArn").unwrap_or_default();
let policy = param(req, "Policy").unwrap_or_default();
let mut guard = self.state.write();
let acct = guard.account(&req.account_id);
acct.resource_policies.insert(arn.clone(), policy.clone());
Ok(xml_resp(
"PutResourcePolicy",
format!(
"<ResourcePolicy><ResourceArn>{}</ResourceArn><Policy>{}</Policy></ResourcePolicy>",
xml_escape(&arn),
xml_escape(&policy)
),
&req.request_id,
))
}
pub(super) fn get_resource_policy(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let arn = param(req, "ResourceArn").unwrap_or_default();
let guard = self.state.read();
let policy = guard
.accounts
.get(&req.account_id)
.and_then(|a| a.resource_policies.get(&arn))
.cloned()
.ok_or_else(|| resource_not_found(format!("Resource policy for {arn} not found.")))?;
Ok(xml_resp(
"GetResourcePolicy",
format!(
"<ResourcePolicy><ResourceArn>{}</ResourceArn><Policy>{}</Policy></ResourcePolicy>",
xml_escape(&arn),
xml_escape(&policy)
),
&req.request_id,
))
}
pub(super) fn delete_resource_policy(
&self,
req: &AwsRequest,
) -> Result<AwsResponse, AwsServiceError> {
let arn = param(req, "ResourceArn").unwrap_or_default();
let mut guard = self.state.write();
let acct = guard.account(&req.account_id);
if acct.resource_policies.remove(&arn).is_none() {
return Err(resource_not_found(format!(
"Resource policy for {arn} not found."
)));
}
Ok(xml_metadata_only("DeleteResourcePolicy", &req.request_id))
}
pub(super) fn create_tags(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
let arn = param(req, "ResourceName").unwrap_or_default();
let new_tags = parse_tags(req);
let mut guard = self.state.write();
let acct = guard.account(&req.account_id);
let tags = resolve_tags_mut(acct, &arn)
.ok_or_else(|| resource_not_found(format!("Cannot find resource with ARN: {arn}")))?;
for nt in new_tags {
if let Some(existing) = tags.iter_mut().find(|t| t.key == nt.key) {
existing.value = nt.value;
} else {
tags.push(nt);
}
}
Ok(xml_metadata_only("CreateTags", &req.request_id))
}
pub(super) fn delete_tags(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
let arn = param(req, "ResourceName").unwrap_or_default();
let keys = member_list(req, "TagKeys", "TagKey");
let mut guard = self.state.write();
let acct = guard.account(&req.account_id);
let tags = resolve_tags_mut(acct, &arn)
.ok_or_else(|| resource_not_found(format!("Cannot find resource with ARN: {arn}")))?;
tags.retain(|t| !keys.contains(&t.key));
Ok(xml_metadata_only("DeleteTags", &req.request_id))
}
pub(super) fn describe_tags(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
let name_filter = param(req, "ResourceName");
let type_filter = param(req, "ResourceType");
let guard = self.state.read();
let mut rows = String::new();
if let Some(acct) = guard.accounts.get(&req.account_id) {
let region = &req.region;
let account_id = &req.account_id;
let mut push = |resource_type: &str, name: &str, tags: &[crate::state::Tag]| {
let arn = Arn::new(
"redshift",
region,
account_id,
&format!("{resource_type}:{name}"),
)
.to_string();
if name_filter.as_ref().is_some_and(|n| n != &arn) {
return;
}
if type_filter.as_ref().is_some_and(|t| t != resource_type) {
return;
}
for t in tags {
rows.push_str(&format!(
"<TaggedResource><Tag><Key>{}</Key><Value>{}</Value></Tag><ResourceName>{}</ResourceName><ResourceType>{}</ResourceType></TaggedResource>",
xml_escape(&t.key),
xml_escape(&t.value),
xml_escape(&arn),
xml_escape(resource_type)
));
}
};
for c in acct.clusters.values() {
push("cluster", &c.cluster_identifier, &c.tags);
}
for s in acct.snapshots.values() {
push("snapshot", &s.snapshot_identifier, &s.tags);
}
for g in acct.parameter_groups.values() {
push("parametergroup", &g.parameter_group_name, &g.tags);
}
for g in acct.subnet_groups.values() {
push("subnetgroup", &g.cluster_subnet_group_name, &g.tags);
}
for g in acct.security_groups.values() {
push("securitygroup", &g.cluster_security_group_name, &g.tags);
}
for g in acct.snapshot_copy_grants.values() {
push("snapshotcopygrant", &g.snapshot_copy_grant_name, &g.tags);
}
for s in acct.snapshot_schedules.values() {
push("snapshotschedule", &s.schedule_identifier, &s.tags);
}
for h in acct.hsm_client_certificates.values() {
push(
"hsmclientcertificate",
&h.hsm_client_certificate_identifier,
&h.tags,
);
}
for h in acct.hsm_configurations.values() {
push("hsmconfiguration", &h.hsm_configuration_identifier, &h.tags);
}
for s in acct.event_subscriptions.values() {
push("eventsubscription", &s.subscription_name, &s.tags);
}
for u in acct.usage_limits.values() {
push("usagelimit", &u.usage_limit_id, &u.tags);
}
}
Ok(xml_resp(
"DescribeTags",
format!("<TaggedResources>{rows}</TaggedResources>"),
&req.request_id,
))
}
}
fn resolve_tags_mut<'a>(
acct: &'a mut crate::state::RedshiftState,
arn: &str,
) -> Option<&'a mut Vec<crate::state::Tag>> {
let parts: Vec<&str> = arn.splitn(7, ':').collect();
if parts.len() < 7 || parts[2] != "redshift" {
return None;
}
let resource_type = parts[5];
let name = parts[6];
match resource_type {
"cluster" => acct.clusters.get_mut(name).map(|c| &mut c.tags),
"snapshot" => {
let snap_id = name.rsplit('/').next().unwrap_or(name);
acct.snapshots.get_mut(snap_id).map(|s| &mut s.tags)
}
"parametergroup" => acct.parameter_groups.get_mut(name).map(|g| &mut g.tags),
"subnetgroup" => acct.subnet_groups.get_mut(name).map(|g| &mut g.tags),
"securitygroup" => acct.security_groups.get_mut(name).map(|g| &mut g.tags),
"snapshotcopygrant" => acct.snapshot_copy_grants.get_mut(name).map(|g| &mut g.tags),
"snapshotschedule" => acct.snapshot_schedules.get_mut(name).map(|s| &mut s.tags),
"hsmclientcertificate" => acct
.hsm_client_certificates
.get_mut(name)
.map(|h| &mut h.tags),
"hsmconfiguration" => acct.hsm_configurations.get_mut(name).map(|h| &mut h.tags),
"eventsubscription" => acct.event_subscriptions.get_mut(name).map(|s| &mut s.tags),
"usagelimit" => acct.usage_limits.get_mut(name).map(|u| &mut u.tags),
_ => None,
}
}
fn extract_target_action(req: &AwsRequest) -> Option<String> {
for variant in ["ResizeCluster", "PauseCluster", "ResumeCluster"] {
let ci_key = format!("TargetAction.{variant}.ClusterIdentifier");
if let Some(ci) = req.query_params.get(&ci_key) {
return Some(format!(
"<{variant}><ClusterIdentifier>{}</ClusterIdentifier></{variant}>",
xml_escape(ci)
));
}
}
None
}