Loading_
Loading_
Finds unattached disks, idle public IPs, empty NICs, stale snapshots and unused NSGs across every subscription, with a monthly cost estimate per finding.
"""Find and price orphaned Azure resources across all accessible subscriptions. Read-only. Produces a CSV of candidates with estimated monthly cost. pip install azure-identity azure-mgmt-resource azure-mgmt-compute \ azure-mgmt-network azure-mgmt-subscription requests python azure_orphan_sweep.py --output orphans.csv""" from __future__ import annotations import argparseimport csvimport loggingfrom dataclasses import dataclass, asdict, field import requestsfrom azure.identity import DefaultAzureCredentialfrom azure.mgmt.compute import ComputeManagementClientfrom azure.mgmt.network import NetworkManagementClientfrom azure.mgmt.subscription import SubscriptionClient RETAIL_PRICES = "https://prices.azure.com/api/retail/prices"log = logging.getLogger("orphan-sweep") @dataclassclass Finding: subscription: str resource_group: str name: str kind: str location: str detail: str monthly_usd: float = 0.0 tags: str = "" class PriceBook: """Tiny memoised wrapper over the public retail price API.""" def __init__(self) -> None: self._cache: dict[tuple[str, str], float] = {} def monthly(self, meter: str, region: str) -> float: key = (meter, region) if key in self._cache: return self._cache[key] price = 0.0 try: query = "serviceName eq 'Storage' and armRegionName eq '" + region + "'" resp = requests.get( RETAIL_PRICES, params={"$filter": query, "$top": 100}, timeout=20, ) resp.raise_for_status() for item in resp.json().get("Items", []): if meter.lower() in item.get("meterName", "").lower(): price = float(item.get("retailPrice", 0.0)) * 730 break except Exception as exc: # pricing is best-effort, never fatal log.debug("price lookup failed for %s/%s: %s", meter, region, exc) self._cache[key] = price return price def sweep_subscription(cred, sub_id: str, sub_name: str, prices: PriceBook) -> list[Finding]: findings: list[Finding] = [] compute = ComputeManagementClient(cred, sub_id) network = NetworkManagementClient(cred, sub_id) def rg_of(resource_id: str) -> str: parts = resource_id.split("/") return parts[4] if len(parts) > 4 else "?" # Unattached managed disks for disk in compute.disks.list(): if disk.disk_state == "Unattached": findings.append(Finding( subscription=sub_name, resource_group=rg_of(disk.id), name=disk.name, kind="Managed disk", location=disk.location, detail=str(disk.disk_size_gb) + " GB " + str(disk.sku.name), monthly_usd=round(prices.monthly("LRS Provisioned", disk.location) * (disk.disk_size_gb or 0), 2), tags=",".join((disk.tags or {}).keys()), )) # Public IPs with no attachment for ip in network.public_ip_addresses.list_all(): if ip.ip_configuration is None: findings.append(Finding( subscription=sub_name, resource_group=rg_of(ip.id), name=ip.name, kind="Public IP", location=ip.location, detail=str(ip.sku.name if ip.sku else "Basic") + " / " + str(ip.public_ip_allocation_method), monthly_usd=3.65, tags=",".join((ip.tags or {}).keys()), )) # NICs attached to nothing for nic in network.network_interfaces.list_all(): if nic.virtual_machine is None and not nic.private_link_service: findings.append(Finding( subscription=sub_name, resource_group=rg_of(nic.id), name=nic.name, kind="Network interface", location=nic.location, detail="No VM attached", tags=",".join((nic.tags or {}).keys()), )) # NSGs bound to no subnet and no NIC for nsg in network.network_security_groups.list_all(): if not nsg.subnets and not nsg.network_interfaces: findings.append(Finding( subscription=sub_name, resource_group=rg_of(nsg.id), name=nsg.name, kind="Network security group", location=nsg.location, detail=str(len(nsg.security_rules or [])) + " custom rules, unbound", )) # Snapshots whose source disk is gone live_disks = {d.id.lower() for d in compute.disks.list()} for snap in compute.snapshots.list(): source = (snap.creation_data.source_resource_id or "").lower() if source and source not in live_disks: findings.append(Finding( subscription=sub_name, resource_group=rg_of(snap.id), name=snap.name, kind="Snapshot", location=snap.location, detail="Source disk no longer exists", monthly_usd=round(0.05 * (snap.disk_size_gb or 0), 2), tags=",".join((snap.tags or {}).keys()), )) return findings def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--output", default="orphans.csv") parser.add_argument("--subscription", action="append", help="Limit to specific IDs") parser.add_argument("--verbose", action="store_true") args = parser.parse_args() logging.basicConfig( level=logging.DEBUG if args.verbose else logging.INFO, format="%(levelname)-7s %(message)s", ) cred = DefaultAzureCredential() prices = PriceBook() subs = SubscriptionClient(cred).subscriptions.list() all_findings: list[Finding] = [] for sub in subs: if args.subscription and sub.subscription_id not in args.subscription: continue if sub.state != "Enabled": continue log.info("Scanning %s", sub.display_name) try: all_findings.extend( sweep_subscription(cred, sub.subscription_id, sub.display_name, prices) ) except Exception as exc: log.error(" failed: %s", exc) all_findings.sort(key=lambda f: f.monthly_usd, reverse=True) with open(args.output, "w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=list(asdict(Finding("", "", "", "", "", "")).keys())) writer.writeheader() for finding in all_findings: writer.writerow(asdict(finding)) total = sum(f.monthly_usd for f in all_findings) log.info("") log.info("%d orphaned resources, about %.2f USD/month", len(all_findings), total) for kind in sorted({f.kind for f in all_findings}): subset = [f for f in all_findings if f.kind == kind] log.info(" %-24s %3d %8.2f USD", kind, len(subset), sum(f.monthly_usd for f in subset)) log.info("Written to %s", args.output) return 0 if __name__ == "__main__": raise SystemExit(main())Orphaned resources are the cheapest cloud saving available and the one nobody has time to hunt. Every deleted VM leaves a trail: the OS disk if delete-on-terminate was off, the NIC, the public IP, and whatever snapshots someone took before the change.
This walks every subscription the credential can see, correlates each candidate against its attachment state, and prices it from the retail rates API so the output is a number a finance team can act on rather than a resource list.
Nothing is deleted. The output is a CSV plus an optional Azure Resource Graph query you can paste into the portal to review the same set interactively.
| Name | Type | Required | Description |
|---|---|---|---|
--output | path | Optional | CSV destination. |
--subscription | string[] | Optional | Limit the sweep to specific subscription IDs. |
The platform turns any script into a governed automation — versioned, gated, audited and reversible.