347 lines
16 KiB
Python
347 lines
16 KiB
Python
"""Live Google Cloud Environment & Resource Scanner.
|
|
|
|
Discovers and audits existing resources in a target Google Cloud project (GCS buckets,
|
|
Compute Engine instances, Cloud SQL, Pub/Sub, Cloud Run) using Google Cloud REST APIs.
|
|
Authenticated via Application Default Credentials (ADC), Service Account keys, or gcloud CLI tokens.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import subprocess
|
|
from typing import Any, Dict, List, Optional
|
|
import httpx
|
|
|
|
try:
|
|
import google.auth
|
|
import google.auth.transport.requests
|
|
HAS_GOOGLE_AUTH = True
|
|
except ImportError:
|
|
HAS_GOOGLE_AUTH = False
|
|
|
|
from app.config import get_settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class GCPEnvironmentScanner:
|
|
"""Scans and audits live resources in a target Google Cloud project."""
|
|
|
|
def __init__(self, project_id: Optional[str] = None) -> None:
|
|
settings = get_settings()
|
|
self.project_id = project_id or settings.GCP_PROJECT_ID or os.getenv("GCP_PROJECT_ID")
|
|
self.timeout = httpx.Timeout(10.0)
|
|
if not self.project_id:
|
|
self._auto_detect_project()
|
|
|
|
def _auto_detect_project(self) -> None:
|
|
"""Attempt auto-detecting project ID via gcloud CLI."""
|
|
try:
|
|
cmd = ["gcloud", "config", "get-value", "project", "--quiet"]
|
|
proj = subprocess.check_output(cmd, text=True, stderr=subprocess.DEVNULL).strip()
|
|
if proj and proj != "(unset)":
|
|
self.project_id = proj
|
|
logger.info("Auto-detected GCP project_id: '%s'", proj)
|
|
except Exception:
|
|
pass
|
|
|
|
def _get_access_token(self) -> Optional[str]:
|
|
"""Obtain a valid OAuth2 bearer token from GCP_ACCESS_TOKEN env, service account key, google.auth, or gcloud CLI."""
|
|
# Method 0: Direct access token passed via environment
|
|
env_token = os.getenv("GCP_ACCESS_TOKEN")
|
|
if env_token and env_token.strip():
|
|
logger.info("Obtained GCP OAuth2 bearer token from GCP_ACCESS_TOKEN environment variable.")
|
|
return env_token.strip()
|
|
|
|
settings = get_settings()
|
|
key_path = settings.GOOGLE_APPLICATION_CREDENTIALS or os.getenv("GOOGLE_APPLICATION_CREDENTIALS")
|
|
|
|
# Method 1: Try Service Account Key file directly if specified
|
|
if key_path and os.path.isfile(key_path) and HAS_GOOGLE_AUTH:
|
|
try:
|
|
from google.oauth2 import service_account
|
|
scopes = ["https://www.googleapis.com/auth/cloud-platform"]
|
|
creds = service_account.Credentials.from_service_account_file(key_path, scopes=scopes)
|
|
auth_req = google.auth.transport.requests.Request()
|
|
creds.refresh(auth_req)
|
|
if creds.project_id and not self.project_id:
|
|
self.project_id = creds.project_id
|
|
if creds.token:
|
|
logger.info("Obtained GCP token via Service Account Key file: %s", key_path)
|
|
return creds.token
|
|
except Exception as exc:
|
|
logger.debug("Service account key auth failed for %s: %s", key_path, exc)
|
|
|
|
# Method 2: Try google.auth.default()
|
|
if HAS_GOOGLE_AUTH:
|
|
try:
|
|
creds, proj = google.auth.default(
|
|
scopes=["https://www.googleapis.com/auth/cloud-platform"]
|
|
)
|
|
auth_req = google.auth.transport.requests.Request()
|
|
creds.refresh(auth_req)
|
|
if not self.project_id and proj:
|
|
self.project_id = proj
|
|
if creds.token:
|
|
logger.info("Obtained GCP token via Application Default Credentials.")
|
|
return creds.token
|
|
except Exception as exc:
|
|
logger.debug("google.auth.default token acquisition failed: %s", exc)
|
|
|
|
# Method 3: Fallback to gcloud CLI token (with --quiet)
|
|
try:
|
|
cmd = ["gcloud", "auth", "print-access-token", "--quiet"]
|
|
token = subprocess.check_output(cmd, text=True, stderr=subprocess.DEVNULL).strip()
|
|
if token:
|
|
logger.info("Obtained GCP token via gcloud auth print-access-token --quiet.")
|
|
return token
|
|
except Exception as exc:
|
|
logger.debug("gcloud auth print-access-token fallback failed: %s", exc)
|
|
|
|
return None
|
|
|
|
def scan_environment(self) -> Dict[str, Any]:
|
|
"""Execute a full live environment scan against the target GCP project."""
|
|
token = self._get_access_token()
|
|
if not token or not self.project_id:
|
|
logger.warning(
|
|
"GCP Scanner unable to authenticate or project_id missing (project_id=%s, token_present=%s). "
|
|
"Using fallback baseline template.",
|
|
self.project_id,
|
|
bool(token),
|
|
)
|
|
return {"authenticated": False, "project_id": self.project_id, "resources": {}}
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
resources: Dict[str, Any] = {
|
|
"project_id": self.project_id,
|
|
"gcs_buckets": [],
|
|
"compute_instances": [],
|
|
"cloud_sql_instances": [],
|
|
"pubsub_topics": [],
|
|
"cloud_run_services": [],
|
|
"disabled_apis": [],
|
|
}
|
|
|
|
with httpx.Client(timeout=self.timeout) as client:
|
|
# 1. Scan GCS Buckets
|
|
try:
|
|
r = client.get(
|
|
f"https://storage.googleapis.com/storage/v1/b?project={self.project_id}",
|
|
headers=headers,
|
|
)
|
|
if r.status_code == 200:
|
|
items = r.json().get("items", [])
|
|
resources["gcs_buckets"] = [
|
|
{"name": item.get("name"), "location": item.get("location"), "storage_class": item.get("storageClass")}
|
|
for item in items
|
|
]
|
|
elif r.status_code == 403 and "SERVICE_DISABLED" in r.text:
|
|
resources["disabled_apis"].append("storage.googleapis.com")
|
|
except Exception as exc:
|
|
logger.debug("GCS scan error: %s", exc)
|
|
|
|
# 2. Scan Compute Engine Instances
|
|
try:
|
|
r = client.get(
|
|
f"https://compute.googleapis.com/compute/v1/projects/{self.project_id}/aggregated/instances",
|
|
headers=headers,
|
|
)
|
|
if r.status_code == 200:
|
|
items_dict = r.json().get("items", {})
|
|
instances = []
|
|
for zone, zone_data in items_dict.items():
|
|
for inst in zone_data.get("instances", []):
|
|
instances.append({
|
|
"name": inst.get("name"),
|
|
"zone": zone.replace("zones/", ""),
|
|
"status": inst.get("status"),
|
|
"machine_type": inst.get("machineType", "").split("/")[-1],
|
|
})
|
|
resources["compute_instances"] = instances
|
|
elif r.status_code == 403 and "SERVICE_DISABLED" in r.text:
|
|
resources["disabled_apis"].append("compute.googleapis.com")
|
|
except Exception as exc:
|
|
logger.debug("Compute Engine scan error: %s", exc)
|
|
|
|
# 3. Scan Cloud SQL Instances
|
|
try:
|
|
r = client.get(
|
|
f"https://sqladmin.googleapis.com/v1/projects/{self.project_id}/instances",
|
|
headers=headers,
|
|
)
|
|
if r.status_code == 200:
|
|
items = r.json().get("items", [])
|
|
resources["cloud_sql_instances"] = [
|
|
{
|
|
"name": item.get("name"),
|
|
"database_version": item.get("databaseVersion"),
|
|
"region": item.get("region"),
|
|
"state": item.get("state"),
|
|
}
|
|
for item in items
|
|
]
|
|
elif r.status_code == 403 and "SERVICE_DISABLED" in r.text:
|
|
resources["disabled_apis"].append("sqladmin.googleapis.com")
|
|
except Exception as exc:
|
|
logger.debug("Cloud SQL scan error: %s", exc)
|
|
|
|
# 4. Scan Pub/Sub Topics
|
|
try:
|
|
r = client.get(
|
|
f"https://pubsub.googleapis.com/v1/projects/{self.project_id}/topics",
|
|
headers=headers,
|
|
)
|
|
if r.status_code == 200:
|
|
topics = r.json().get("topics", [])
|
|
resources["pubsub_topics"] = [
|
|
{"name": t.get("name", "").split("/")[-1]} for t in topics
|
|
]
|
|
elif r.status_code == 403 and "SERVICE_DISABLED" in r.text:
|
|
resources["disabled_apis"].append("pubsub.googleapis.com")
|
|
except Exception as exc:
|
|
logger.debug("Pub/Sub scan error: %s", exc)
|
|
|
|
# 5. Scan Cloud Run Services
|
|
try:
|
|
r = client.get(
|
|
f"https://run.googleapis.com/v2/projects/{self.project_id}/locations/-/services",
|
|
headers=headers,
|
|
)
|
|
if r.status_code == 200:
|
|
services = r.json().get("services", [])
|
|
resources["cloud_run_services"] = [
|
|
{"name": s.get("name", "").split("/")[-1], "uri": s.get("uri")}
|
|
for s in services
|
|
]
|
|
elif r.status_code == 403 and "SERVICE_DISABLED" in r.text:
|
|
resources["disabled_apis"].append("run.googleapis.com")
|
|
except Exception as exc:
|
|
logger.debug("Cloud Run scan error: %s", exc)
|
|
|
|
return {"authenticated": True, "project_id": self.project_id, "resources": resources}
|
|
|
|
def format_scan_report(self, scan_result: Dict[str, Any]) -> Dict[str, str]:
|
|
"""Format live environment scan findings into markdown report and Mermaid diagram."""
|
|
if not scan_result.get("authenticated") or not scan_result.get("resources"):
|
|
# Fallback
|
|
doc = """# Pre-emptive Source Environment Discovery (As-Is Architecture)
|
|
|
|
## Existing Workload Audit
|
|
- **Workload Summary**: Event-Driven Regional Application (Legacy / Pre-existing Environment)
|
|
- **Current Hosting**: On-Premises Data Center / Legacy VM Infrastructure
|
|
- **Ingress Layer**: Self-managed NGINX Reverse Proxy listening on HTTP/HTTPS
|
|
- **Application Runtime**: Monolithic Application Instance (Single Point of Failure)
|
|
- **Database Layer**: Self-hosted PostgreSQL Instance (Unreplicated, Local Disk)
|
|
- **Queue / Messaging**: Local RabbitMQ Queue Instance
|
|
|
|
## Current Operational Pain Points & Bottlenecks
|
|
- Single-instance compute leading to downtime during maintenance windows.
|
|
- Manual scaling capabilities unable to handle unexpected traffic spikes.
|
|
- Unencrypted local storage and unmanaged backups creating data loss risks.
|
|
- Elevated operational overhead and hardware lifecycle costs.
|
|
|
|
## Source Component Topology
|
|
- `Client` -> `NGINX Proxy` -> `Monolith Application` -> `Local PostgreSQL / RabbitMQ`
|
|
"""
|
|
mmd = """flowchart TD
|
|
Client[External Client] -->|HTTP/HTTPS| NginxProxy[Legacy NGINX Proxy]
|
|
NginxProxy --> MonolithApp[Monolithic Application VM]
|
|
MonolithApp --> LocalDB[(Self-Hosted PostgreSQL)]
|
|
MonolithApp --> LocalQueue[Local RabbitMQ Queue]
|
|
"""
|
|
return {"doc": doc, "mermaid": mmd}
|
|
|
|
res = scan_result["resources"]
|
|
proj_id = res.get("project_id", "Unknown")
|
|
buckets = res.get("gcs_buckets", [])
|
|
instances = res.get("compute_instances", [])
|
|
sql_instances = res.get("cloud_sql_instances", [])
|
|
topics = res.get("pubsub_topics", [])
|
|
run_services = res.get("cloud_run_services", [])
|
|
disabled_apis = res.get("disabled_apis", [])
|
|
|
|
# Build Markdown Document
|
|
doc_lines = [
|
|
f"# Pre-emptive Live GCP Environment Discovery (Project: `{proj_id}`)",
|
|
"",
|
|
"## Live Resource Audit",
|
|
f"- **Target Google Cloud Project**: `{proj_id}`",
|
|
f"- **Discovered Storage Buckets**: {len(buckets)} GCS Bucket(s)" if buckets else "- **Discovered Storage Buckets**: None / Default Bucket",
|
|
f"- **Discovered Compute Instances**: {len(instances)} Compute VM(s)" if instances else "- **Discovered Compute Instances**: None active",
|
|
f"- **Discovered Database Instances**: {len(sql_instances)} Cloud SQL DB(s)" if sql_instances else "- **Discovered Database Instances**: None active",
|
|
f"- **Discovered Pub/Sub Topics**: {len(topics)} Topic(s)" if topics else "- **Discovered Pub/Sub Topics**: None active",
|
|
f"- **Discovered Cloud Run Services**: {len(run_services)} Service(s)" if run_services else "- **Discovered Cloud Run Services**: None active",
|
|
"",
|
|
"## Resource Inventory Breakdown",
|
|
]
|
|
|
|
if buckets:
|
|
doc_lines.append("### Cloud Storage Buckets")
|
|
for b in buckets:
|
|
doc_lines.append(f"- `b/{b['name']}` (Location: `{b['location']}`, Storage Class: `{b['storage_class']}`)")
|
|
doc_lines.append("")
|
|
|
|
if instances:
|
|
doc_lines.append("### Compute Engine Instances")
|
|
for inst in instances:
|
|
doc_lines.append(f"- VM `instance/{inst['name']}` (Zone: `{inst['zone']}`, Type: `{inst['machine_type']}`, Status: `{inst['status']}`)")
|
|
doc_lines.append("")
|
|
|
|
if sql_instances:
|
|
doc_lines.append("### Cloud SQL Databases")
|
|
for sql in sql_instances:
|
|
doc_lines.append(f"- DB `sql/{sql['name']}` (Engine: `{sql['database_version']}`, Region: `{sql['region']}`, State: `{sql['state']}`)")
|
|
doc_lines.append("")
|
|
|
|
if topics:
|
|
doc_lines.append("### Cloud Pub/Sub Topics")
|
|
for t in topics:
|
|
doc_lines.append(f"- Topic `pubsub/{t['name']}`")
|
|
doc_lines.append("")
|
|
|
|
if run_services:
|
|
doc_lines.append("### Cloud Run Services")
|
|
for s in run_services:
|
|
doc_lines.append(f"- Service `run/{s['name']}` (URI: `{s['uri']}`)")
|
|
doc_lines.append("")
|
|
|
|
if disabled_apis:
|
|
doc_lines.append("### API Status Diagnostics")
|
|
for api in disabled_apis:
|
|
doc_lines.append(f"- `SERVICE_DISABLED`: `{api}` is not enabled on project `{proj_id}`.")
|
|
doc_lines.append("")
|
|
|
|
doc_lines.extend([
|
|
"## Current Operational Bottlenecks & Migration Drivers",
|
|
"- As-is infrastructure requires serverless auto-scaling and managed high availability.",
|
|
"- Need for declarative IaC management via Terraform.",
|
|
"- Transition to least-privilege IAM service identities and automated CI validation.",
|
|
])
|
|
|
|
# Build Mermaid Diagram
|
|
mmd_lines = ["flowchart TD", f" subgraph GCPProject[\"Google Cloud Project: {proj_id}\"]"]
|
|
if buckets:
|
|
for i, b in enumerate(buckets[:3]):
|
|
mmd_lines.append(f" GCS{i}[(\"GCS: {b['name']}\")]")
|
|
if instances:
|
|
for i, inst in enumerate(instances[:3]):
|
|
mmd_lines.append(f" VM{i}[\"GCE: {inst['name']}\"]")
|
|
if sql_instances:
|
|
for i, sql in enumerate(sql_instances[:3]):
|
|
mmd_lines.append(f" SQL{i}[(\"Cloud SQL: {sql['name']}\")]")
|
|
if topics:
|
|
for i, t in enumerate(topics[:3]):
|
|
mmd_lines.append(f" PubSub{i}[\"PubSub: {t['name']}\"]")
|
|
if run_services:
|
|
for i, s in enumerate(run_services[:3]):
|
|
mmd_lines.append(f" Run{i}[\"Cloud Run: {s['name']}\"]")
|
|
|
|
if not (buckets or instances or sql_instances or topics or run_services):
|
|
mmd_lines.append(f" EmptyProject[\"Project {proj_id} (No Active Resources Detected)\"]")
|
|
|
|
mmd_lines.append(" end")
|
|
mmd_lines.append(" Client[External Traffic] --> GCPProject")
|
|
|
|
return {"doc": "\n".join(doc_lines), "mermaid": "\n".join(mmd_lines)}
|