sb-send.py

#!/usr/bin/env python3
"""Send a deploy envelope to an Azure Service Bus queue from CI (stdlib only, no SDK).

Usage (e.g. an Azure DevOps YAML step, where PublishToAzureServiceBus@1 is not available on agents):

    python3 sb-send.py --namespace mybus --queue deploy-this \
        --policy send-only --key "$SB_KEY" --app hello --ref "$BUILD_SOURCEVERSION" \
        --requested-by "ado:$BUILD_BUILDID"

Uses the Service Bus REST API with a SAS token (https://learn.microsoft.com/azure/service-bus-messaging/service-bus-sas).
"""

from __future__ import annotations

import argparse
import base64
import hashlib
import hmac
import json
import os
import sys
import time
import urllib.parse
import urllib.request
import uuid


def sign_envelope(secret: str, envelope: dict) -> str:
    """HMAC-SHA256 over the fields that decide what gets deployed (deploy_this.triggers.signing)."""
    parts = [
        "deploy-this-envelope-v1",
        envelope["app"],
        envelope["action"],
        envelope.get("ref") or "",
        envelope.get("image") or "",
        envelope["requested_by"],
        envelope["message_id"],
        str(envelope["issued_at"]),
        envelope.get("instance") or "",
        envelope.get("ttl") or "",
    ]
    return hmac.new(secret.encode(), "\n".join(parts).encode(), hashlib.sha256).hexdigest()


def sas_token(uri: str, policy: str, key: str, ttl: int = 300) -> str:
    expiry = int(time.time()) + ttl
    encoded_uri = urllib.parse.quote_plus(uri)
    to_sign = f"{encoded_uri}\n{expiry}".encode()
    signature = base64.b64encode(hmac.new(key.encode(), to_sign, hashlib.sha256).digest()).decode()
    return f"SharedAccessSignature sr={encoded_uri}&sig={urllib.parse.quote_plus(signature)}&se={expiry}&skn={policy}"


def main() -> int:
    ap = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    ap.add_argument(
        "--namespace",
        required=True,
        help="Service Bus namespace name (without .servicebus.windows.net)",
    )
    ap.add_argument("--queue", required=True)
    ap.add_argument(
        "--policy", required=True, help="shared access policy name (Send claim is enough)"
    )
    ap.add_argument(
        "--key",
        default=os.environ.get("SB_KEY"),
        help="shared access key (or $SB_KEY; prefer the environment so it stays out of `ps` and shell history)",
    )
    ap.add_argument("--app", required=True)
    ap.add_argument("--ref", help="commit SHA (preferred), branch or tag -- the box builds it")
    ap.add_argument(
        "--image", help="image built in CI: <repository>@sha256:... -- the box pulls it"
    )
    ap.add_argument(
        "--action",
        default="apply",
        choices=["apply", "stage", "swap", "unstage", "rollback", "remove"],
    )
    ap.add_argument("--instance", help="preview instance name, e.g. pr-123 (apply/remove)")
    ap.add_argument("--ttl", help="preview lifetime, e.g. 3d")
    ap.add_argument("--requested-by", default="ci")
    ap.add_argument(
        "--secret",
        default=os.environ.get("DEPLOY_THIS_ENVELOPE_SECRET"),
        help="HMAC secret shared with the box (or $DEPLOY_THIS_ENVELOPE_SECRET); signs the envelope",
    )
    args = ap.parse_args()
    if not args.key:
        ap.error("--key or $SB_KEY is required")
    if not args.ref and not args.image and args.action in ("apply", "stage"):
        ap.error("one of --ref or --image is required")

    uri = f"https://{args.namespace}.servicebus.windows.net/{args.queue}"
    message_id = str(uuid.uuid4())
    envelope = {
        "app": args.app,
        "action": args.action,
        "requested_by": args.requested_by,
        "message_id": message_id,
    }
    if args.ref:
        envelope["ref"] = args.ref
    if args.image:
        envelope["image"] = args.image
    if args.instance:
        envelope["instance"] = args.instance
    if args.ttl:
        envelope["ttl"] = args.ttl
    if args.secret:
        envelope["issued_at"] = int(time.time())
        envelope["signature"] = sign_envelope(args.secret, envelope)
    body = json.dumps(envelope).encode()
    req = urllib.request.Request(
        f"{uri}/messages",
        data=body,
        method="POST",
        headers={
            "Authorization": sas_token(uri, args.policy, args.key),
            "Content-Type": "application/json",
            "BrokerProperties": json.dumps({"MessageId": message_id}),
        },
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        print(
            f"sent {args.action} {args.app} ({args.image or args.ref}) as message {message_id}: HTTP {resp.status}"
        )
    return 0


if __name__ == "__main__":
    sys.exit(main())