scripts/b2_create_bucket.py

#!/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())