#!/usr/bin/env python3 """Create a Backblaze B2 bucket via the **Native API** (``b2_create_bucket``). Uses credentials from parser ``secrets.toml`` ``[b2]`` (``key_id`` + ``application_key``) by default. Those keys are often **restricted to one bucket** and **cannot** create buckets (no ``writeBuckets``). If authorize shows no ``writeBuckets``, set one of: - ``B2_ADMIN_KEY_ID`` + ``B2_ADMIN_APPLICATION_KEY`` — e.g. **Master Application Key** from the B2 UI (master works for Native API; it still cannot be used for S3 uploads — use a normal app key on the new bucket for ``[b2]`` afterward). Then re-run this script **without** admin env to use ``[b2]`` only, or pass ``--admin`` to use admin env for create only. After success, update ``secrets.toml`` ``[b2].bucket_name`` (and ``region`` from printed ``s3`` URL), create a **new Application Key** scoped to that bucket for S3, and replace ``[b2]`` key_id/application_key. Example:: export B2_ADMIN_KEY_ID=... export B2_ADMIN_APPLICATION_KEY=... python scripts/b2_create_bucket.py --name kink-catalog-ronheichman Docs: https://www.backblaze.com/apidocs/b2-create-bucket """ from __future__ import annotations import argparse import base64 import json import os import re import sys from pathlib import Path from backend.b2_toml import load_b2_section, resolve_b2_config_path ROOT = Path(__file__).resolve().parent.parent DEFAULT_SECRETS = Path(os.environ.get("KINK_PARSER_SECRETS_TOML", Path.home() / "PycharmProjects/parser/secrets.toml")) def _basic_auth(key_id: str, app_key: str) -> str: raw = f"{key_id}:{app_key}".encode() return "Basic " + base64.b64encode(raw).decode() def _authorize_v4(key_id: str, app_key: str) -> dict: import httpx r = httpx.get( "https://api.backblazeb2.com/b2api/v4/b2_authorize_account", headers={"Authorization": _basic_auth(key_id, app_key)}, timeout=60.0, ) if r.status_code != 200: raise SystemExit(f"b2_authorize_account: HTTP {r.status_code} {r.text[:500]}") return r.json() def _region_from_s3_url(s3_url: str) -> str: # https://s3.us-east-005.backblazeb2.com -> us-east-005 m = re.search(r"https://s3\.([a-z0-9-]+)\.backblazeb2\.com", s3_url, re.I) if m: return m.group(1).lower() return "" def main() -> int: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--secrets-toml", type=Path, default=DEFAULT_SECRETS, help="Fallback TOML with [b2]") ap.add_argument("--b2-secrets-toml", type=Path, default=None, help="TOML with only [b2] (overrides env and --secrets-toml)") ap.add_argument( "--name", required=True, help="Globally unique bucket name (6–63 chars, letters/digits/-; cannot start with b2-)", ) ap.add_argument( "--type", choices=("allPrivate", "allPublic"), default="allPrivate", help="Bucket visibility (default: allPrivate)", ) ap.add_argument( "--admin", action="store_true", help="Use B2_ADMIN_KEY_ID / B2_ADMIN_APPLICATION_KEY instead of [b2] for authorize", ) ap.add_argument( "--dry-run", action="store_true", help="Authorize and print capabilities only; do not create", ) args = ap.parse_args() if args.admin: kid = (os.environ.get("B2_ADMIN_KEY_ID") or "").strip() akey = (os.environ.get("B2_ADMIN_APPLICATION_KEY") or "").strip() if not kid or not akey: print("b2_create_bucket: --admin requires B2_ADMIN_KEY_ID and B2_ADMIN_APPLICATION_KEY", file=sys.stderr) return 1 else: b2_path = resolve_b2_config_path(cli_b2=args.b2_secrets_toml, cli_fallback=args.secrets_toml) if not b2_path.is_file(): print(f"b2_create_bucket: missing {b2_path}", file=sys.stderr) return 1 try: b2 = load_b2_section(b2_path) except SystemExit as e: print(f"b2_create_bucket: {e}", file=sys.stderr) return 1 kid = b2["key_id"] akey = b2["application_key"] j = _authorize_v4(kid, akey) storage = (j.get("apiInfo") or {}).get("storageApi") or {} allowed = storage.get("allowed") or {} caps = set(allowed.get("capabilities") or []) api_url = (storage.get("apiUrl") or "").rstrip("/") s3_url = (storage.get("s3ApiUrl") or "").strip() region = _region_from_s3_url(s3_url) account_id = j.get("accountId") or "" auth_token = j.get("authorizationToken") or "" print("b2_create_bucket: authorized.", file=sys.stderr) print(f" accountId={account_id[:8]}… apiUrl={api_url}", file=sys.stderr) print(f" s3ApiUrl={s3_url} → region `{region}`" if region else f" s3ApiUrl={s3_url}", file=sys.stderr) print(f" capabilities: {sorted(caps)}", file=sys.stderr) if args.dry_run: ok = "writeBuckets" in caps print( f"b2_create_bucket: --dry-run — would {'be able to' if ok else 'NOT be able to'} create a bucket with this key.", file=sys.stderr, ) if not ok: print( " Fix: export B2_ADMIN_KEY_ID=… B2_ADMIN_APPLICATION_KEY=… (master key) then " "`python scripts/b2_create_bucket.py --admin --dry-run --name x`", file=sys.stderr, ) return 0 if "writeBuckets" not in caps: print( "\nb2_create_bucket: this key cannot call b2_create_bucket (need writeBuckets).\n" " Use Master Application Key once:\n" " export B2_ADMIN_KEY_ID=… B2_ADMIN_APPLICATION_KEY=…\n" " python scripts/b2_create_bucket.py --admin --name YOUR_BUCKET\n" " Then create a normal Application Key on that bucket for S3 and update secrets.toml [b2].", file=sys.stderr, ) return 1 import httpx url = f"{api_url}/b2api/v4/b2_create_bucket" body = {"accountId": account_id, "bucketName": args.name, "bucketType": args.type} r = httpx.post( url, headers={"Authorization": auth_token}, json=body, timeout=60.0, ) if r.status_code != 200: try: err = r.json() msg = err.get("message", r.text) code = err.get("code", "") except Exception: msg, code = r.text[:800], "" print(f"b2_create_bucket: failed HTTP {r.status_code} {code}: {msg}", file=sys.stderr) return 1 out = r.json() print(json.dumps(out, indent=2)) print( f"\nb2_create_bucket: OK — bucket `{out.get('bucketName')}` id={out.get('bucketId')}", file=sys.stderr, ) print( "Next: create an Application Key restricted to this bucket (read+write files, listAllBucketNames for S3). " "Update parser secrets.toml [b2]:", file=sys.stderr, ) print(f" bucket_name = \"{args.name}\"", file=sys.stderr) if region: print(f" region = \"{region}\"", file=sys.stderr) return 0 if __name__ == "__main__": raise SystemExit(main())