🔱 Imperial Deployment: Shadow Brain Core ignition

VERSION

@@ -1 +1 @@
-v2.9.67
+v2.9.71

scripts/tools/cloud_tool.py

@@ -71,32 +71,39 @@ import time as _time
 _R2_LIST_CACHE: dict = {}  # key: (bucket, path) → {'files': [...], 'ts': float}
 _R2_LIST_TTL = 30  # 초
 
+# 🔱 [Imperial Cloud Guard] Determine if running in cloud
+IS_RENDER = os.environ.get('RENDER') == 'true'
+IS_CLOUD_RUN = os.environ.get('K_SERVICE') is not None
+IS_CLOUD = IS_RENDER or IS_CLOUD_RUN
+
 def _get_bw_credential(name):
-    """Fetches a credential from Bitwarden, trying both spaced and non-spaced Imperial formats.
-    결과는 인메모리 캐시(_BW_CREDENTIAL_CACHE)에 저장되어 반복 호출 시 BW CLI 재실행을 방지합니다.
+    """Fetches a credential. On Cloud (Render/GCP), skips BW and uses Env only.
+    결과는 인메모리 캐시(_BW_CREDENTIAL_CACHE)에 저장됩니다.
     """
-    # If it's already a value (not starting with Imperial:), return as is
     if not isinstance(name, str) or "Imperial:" not in str(name):
         return name
 
-    # 🔱 캐시 히트: 이미 인출한 자격증명은 즉시 반환 (BW CLI 생략)
     if name in _BW_CREDENTIAL_CACHE:
         return _BW_CREDENTIAL_CACHE[name]
 
-    # Try environment variable first (Aegis runner might have injected it)
-    # Strip prefix for env lookup: Imperial:R2_KEY -> R2_KEY
+    # Standardize key for Env lookup
     env_key = name.replace("Imperial:", "").replace("Imperial: ", "").strip()
     env_val = os.environ.get(env_key)
+    
     if env_val and "Imperial:" not in str(env_val):
         _BW_CREDENTIAL_CACHE[name] = env_val
         return env_val
 
+    # 🔱 Cloud Environment: BW CLI is not available.
+    if IS_CLOUD:
+        print(f"[🔱 Cloud Alert] Missing Environment Variable: {env_key}")
+        return None
+
     # Standardize both ways: 'Imperial: KEY' and 'Imperial:KEY'
     name_no_space = name.replace("Imperial: ", "Imperial:").strip()
     name_with_space = name_no_space.replace("Imperial:", "Imperial: ").strip()
 
     candidates = [name, name_no_space, name_with_space]
-    # Remove duplicates while preserving order
     unique_candidates = []
     for c in candidates:
         if c not in unique_candidates:
@@ -104,22 +111,20 @@ def _get_bw_credential(name):
 
     for cand in unique_candidates:
         try:
-            # Use shell=True for Windows to better find 'bw' in PATH
             result = subprocess.run(
                 f'bw get password "{cand}" --raw',
                 capture_output=True, text=True, check=True, shell=True, timeout=30
             )
             val = result.stdout.strip()
             if val:
-                # 🔱 캐시에 저장: 모든 변형 키를 동일 값으로 캐싱
                 for key in unique_candidates:
                     _BW_CREDENTIAL_CACHE[key] = val
                 return val
         except:
             continue
 
-    print(f"[Aegis Error] Failed to fetch {name} (or variations) from BW or Env. (Check Render Environment Variables)")
-    return None # Return None instead of the key name to avoid downstream auth errors
+    print(f"[Aegis Error] Failed to fetch {name} from BW/Env. (Key: {env_key})")
+    return None
 
 
 def clear_bw_credential_cache():
@@ -132,21 +137,17 @@ def clear_bw_credential_cache():
 
 
 def _prewarm_credentials():
-    """🔱 서버 기동 시 모든 R2 계정의 자격증명을 선제 로드합니다.
-    이 함수가 완료된 이후에는 모든 R2 요청이 BW CLI 없이 즉시 응답합니다.
-    """
+    """🔱 서버 기동 시 모든 R2 계정의 자격증명을 선제 로드합니다."""
+    # Cloud에서는 어차피 Env에서 가져오므로 Pre-warm이 불필요하거나 에러 메시지만 남발함
+    if IS_CLOUD:
+        return
+
     all_keys = set()
     for acc in R2_ACCOUNTS.values():
-        key_id = acc.get("access_key_id", "")
-        secret_id = acc.get("secret_access_key", "")
-        acc_id = acc.get("account_id", "")
-        
-        if key_id and "Imperial:" in key_id:
-            all_keys.add(key_id)
-        if secret_id and "Imperial:" in secret_id:
-            all_keys.add(secret_id)
-        if acc_id and "Imperial:" in acc_id:
-            all_keys.add(acc_id)
+        for k in ["access_key_id", "secret_access_key", "account_id"]:
+            key_val = acc.get(k, "")
+            if key_val and "Imperial:" in str(key_val):
+                all_keys.add(key_val)
 
     loaded = 0
     for key in all_keys:
@@ -154,11 +155,14 @@ def _prewarm_credentials():
             val = _get_bw_credential(key)
             if val and "Imperial:" not in str(val):
                 loaded += 1
-    print(f"[Aegis Pre-warm] R2 자격증명 {loaded}/{len(all_keys)}개 선제 로드 완료.")
+    if loaded > 0:
+        print(f"[Aegis Pre-warm] R2 자격증명 {loaded}/{len(all_keys)}개 선제 로드 완료.")
 
 
-# 🔥 서버/모듈 기동 시 자동 Pre-warm (백그라운드 스레드로 실행)
+# 🔥 서버/모듈 기동 시 자동 Pre-warm (로컬 전용)
 def _prewarm_async():
+    if IS_CLOUD:
+        return
     import threading
     t = threading.Thread(target=_prewarm_credentials, name="aegis-prewarm", daemon=True)
     t.start()
@@ -346,11 +350,22 @@ def get_r2_bucket_size(account_config, bucket_name="cache"):
         access_key = _get_bw_credential(account_config.get("access_key_id"))
         secret_key = _get_bw_credential(account_config.get("secret_access_key"))
         account_id = _get_bw_credential(account_config.get("account_id"))
+        
+        if not all([access_key, secret_key, account_id]):
+            missing = []
+            if not access_key: missing.append("ACCESS_KEY_ID")
+            if not secret_key: missing.append("SECRET_ACCESS_KEY")
+            if not account_id: missing.append("ACCOUNT_ID")
+            raise RuntimeError(f"Missing R2 Credentials in Environment: {', '.join(missing)}")
+
+        from botocore.config import Config
         s3 = boto3.client(
             service_name='s3',
             endpoint_url=f'https://{account_id}.r2.cloudflarestorage.com',
             aws_access_key_id=access_key,
             aws_secret_access_key=secret_key,
+            region_name='auto',
+            config=Config(signature_version='s3v4')
         )
         total_bytes = 0
         total_count = 0
@@ -360,7 +375,8 @@ def get_r2_bucket_size(account_config, bucket_name="cache"):
                 if not obj["Key"].endswith("/"):  # 폴더 자체 제외
                     total_bytes += obj["Size"]
                     total_count += 1
-        return {"count": total_count, "bytes": total_bytes}
+        formatted = f"{total_bytes / (1024*1024):.2f} MB" if total_bytes < 1024**3 else f"{total_bytes / (1024**3):.2f} GB"
+        return {"count": total_count, "bytes": total_bytes, "formatted": formatted}
     except Exception as e:
         raise RuntimeError(f"get_r2_bucket_size failed: {e}")