AWS: hardening rules - part 2

cleancloud.yaml

@@ -61,9 +61,8 @@ rules:
 
   aws.rds.instance.idle:
     enabled: true
-    min_cost: 100               # suppress RDS findings below $100/month estimated cost
     params:
-      idle_days: 21             # require 21 days idle before flagging (default: 14)
+      idle_days_threshold: 21   # require 21 days idle before flagging (default: 14)
 
   gcp.sql.instance.idle:
     enabled: true

cleancloud/providers/aws/rules/ai/ec2_gpu_idle.py

@@ -397,7 +397,7 @@ def _list_gpu_metrics(cloudwatch, instance_id: str) -> list:
             Dimensions=[{"Name": "InstanceId", "Value": instance_id}],
         )
         return resp.get("Metrics", [])
-    except ClientError:
+    except Exception:
         return []
 
 
@@ -434,7 +434,7 @@ def _get_max_gpu_utilisation(
             gpu_max = max(dp["Maximum"] for dp in datapoints)
             if max_util is None or gpu_max > max_util:
                 max_util = gpu_max
-        except ClientError:
+        except Exception:
             continue
 
     return max_util
@@ -467,5 +467,5 @@ def _get_avg_cpu_utilisation(
         if not datapoints:
             return None
         return max(dp["Maximum"] for dp in datapoints)
-    except ClientError:
+    except Exception:
         return None

cleancloud/providers/aws/rules/ai/sagemaker_endpoint_idle.py

@@ -434,7 +434,7 @@ def _check_invocations(
                 queried_with_variants=False,
                 fetch_failed=False,
             )
-        except ClientError:
+        except Exception:
             return InvocationCheckResult(
                 has_traffic=True,
                 active_variants=[],
@@ -471,7 +471,7 @@ def _check_invocations(
             else:
                 idle_variants.append(variant_name)
 
-        except ClientError:
+        except Exception:
             # CloudWatch API failure — treat this variant as active and surface the failure.
             return InvocationCheckResult(
                 has_traffic=True,
@@ -561,7 +561,7 @@ def _describe_endpoint(
                 slcfg = cv.get("ServerlessConfig")
                 if slcfg:
                     serverless_cfg_by_variant[cv["VariantName"]] = slcfg
-        except ClientError:
+        except Exception:
             pass  # config inaccessible — costs/GPU will use defaults
 
         accumulated_cost = 0.0
@@ -631,7 +631,7 @@ def _describe_endpoint(
             total_provisioned_concurrency,
         )
 
-    except ClientError:
+    except Exception:
         # Unknown state — return zero instances so the endpoint is skipped rather
         # than flagged with assumed cost and instance count.
         return None, False, 0, 0, None, [], 0

cleancloud/providers/aws/rules/ami_old.py

@@ -507,7 +507,7 @@ def _get_last_launched_time(ec2, ami_id: str) -> Tuple[Optional[datetime], bool]
         if not isinstance(value, str) or not value:
             return None, False
         return datetime.fromisoformat(value.replace("Z", "+00:00")), False
-    except ClientError:
+    except Exception:
         return None, True
 
 
@@ -527,7 +527,7 @@ def _check_active_instances(ec2, ami_id: str) -> Tuple[bool, bool]:
         )
         found = any(r.get("Instances") for r in resp.get("Reservations", []))
         return found, False
-    except ClientError:
+    except Exception:
         return False, True
 
 
@@ -572,11 +572,11 @@ def _build_lt_index(ec2) -> Tuple[Dict[str, List[str]], bool]:
                     v_lt_id = v.get("LaunchTemplateId")
                     if image_id and v_lt_id:
                         index.setdefault(image_id, set()).add(v_lt_id)
-            except ClientError:
+            except Exception:
                 continue  # best-effort per LT
 
         return {k: sorted(v) for k, v in index.items()}, lt_truncated
-    except ClientError:
+    except Exception:
         return {}, True
 
 
@@ -610,5 +610,5 @@ def _build_lc_index(autoscaling) -> Tuple[Dict[str, List[str]], bool]:
                 break
             kwargs["NextToken"] = nxt
         return {k: sorted(v) for k, v in index.items()}, lc_truncated
-    except ClientError:
+    except Exception:
         return {}, True

cleancloud/providers/aws/rules/ebs_snapshot_old.py

@@ -37,7 +37,6 @@
 from typing import List, Optional, Set, Tuple
 
 import boto3
-from botocore.exceptions import BotoCoreError, ClientError
 
 from cleancloud.core.confidence import ConfidenceLevel
 from cleancloud.core.evidence import Evidence
@@ -68,7 +67,7 @@ def _build_ami_snapshot_index(ec2) -> Tuple[Set[str], bool]:
                     snap_id = bdm.get("Ebs", {}).get("SnapshotId")
                     if snap_id:
                         referenced.add(snap_id)
-    except (ClientError, BotoCoreError):
+    except Exception:
         return referenced, True
     return referenced, False
 
@@ -91,7 +90,7 @@ def _check_external_sharing(ec2, snap_id: str) -> Tuple[bool, bool]:
             if perm.get("UserId"):  # explicit cross-account
                 return True, False
         return False, False
-    except (ClientError, BotoCoreError):
+    except Exception:
         return False, True
 
 

cleancloud/providers/aws/rules/ec2_sg_unused.py

@@ -262,7 +262,7 @@ def find_unused_security_groups(
                 )
                 if name:
                     vpc_names[vpc["VpcId"]] = name
-        except (ClientError, BotoCoreError):
+        except Exception:
             pass  # VPC names are display-only; don't fail the rule
 
     # --- Step 7: Apply exclusion rules and emit findings ---

cleancloud/providers/aws/rules/ec2_stopped.py

@@ -331,7 +331,7 @@ def _get_volume_sizes(ec2, volume_ids: List[str]) -> Dict[str, int]:
                 size = vol.get("Size")
                 if vid and size is not None:
                     sizes[vid] = size
-    except (ClientError, BotoCoreError):
+    except Exception:
         pass
     return sizes