Add GCS HMAC key-pair auth as an alternative to service-account JSON

CHANGELOG.md

@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Added
+
+- `define_gcs_connection` now accepts an HMAC key pair (`hmac_access_id`/`hmac_secret`) as an alternative to `service_account_credentials_json`, matching the Forward CLI's `gcloud_storage_hmac`/`gcloud_storage_sa` distinction. Emitted as `GCS_HMAC_ACCESS_ID`/`GCS_HMAC_SECRET` in the generated `.connection` file and round-tripped by the datafile parser and migration emitter.
+
 ## [0.4.0] - 2026-06-29
 
 ### Added

README.md

@@ -550,6 +550,15 @@ landing_gcs = define_gcs_connection(
     },
 )
 
+# Or authenticate with an HMAC key pair instead of a service account:
+landing_gcs_hmac = define_gcs_connection(
+    "landing_gcs_hmac",
+    {
+        "hmac_access_id": secret("GCS_HMAC_ACCESS_ID"),
+        "hmac_secret": secret("GCS_HMAC_SECRET"),
+    },
+)
+
 events_dynamodb = define_dynamodb_connection(
     "events_dynamodb",
     {

src/tinybird_sdk/generator/connection.py

@@ -63,10 +63,16 @@ def _generate_s3_connection(connection: S3ConnectionDefinition) -> str:
 
 def _generate_gcs_connection(connection: GCSConnectionDefinition) -> str:
     options = connection.options
-    parts = [
-        "TYPE gcs",
-        f"GCS_SERVICE_ACCOUNT_CREDENTIALS_JSON {options.service_account_credentials_json}",
-    ]
+    parts = ["TYPE gcs"]
+
+    if options.service_account_credentials_json:
+        parts.append(
+            f"GCS_SERVICE_ACCOUNT_CREDENTIALS_JSON {options.service_account_credentials_json}"
+        )
+    if options.hmac_access_id:
+        parts.append(f"GCS_HMAC_ACCESS_ID {options.hmac_access_id}")
+    if options.hmac_secret:
+        parts.append(f"GCS_HMAC_SECRET {options.hmac_secret}")
 
     return "\n".join(parts)
 

src/tinybird_sdk/migrate/emit_ts.py

@@ -341,9 +341,15 @@ def _emit_gcs_connection(connection: GCSConnectionModel) -> str:
     variable_name = to_snake_case(connection.name)
     lines: list[str] = []
     lines.append(f"{variable_name} = define_gcs_connection({_escape_string(connection.name)}, {{")
-    lines.append(
-        f"    'service_account_credentials_json': {_escape_string(connection.service_account_credentials_json)},"
-    )
+    if connection.service_account_credentials_json:
+        lines.append(
+            f"    'service_account_credentials_json': "
+            f"{_escape_string(connection.service_account_credentials_json)},"
+        )
+    if connection.hmac_access_id:
+        lines.append(f"    'hmac_access_id': {_escape_string(connection.hmac_access_id)},")
+    if connection.hmac_secret:
+        lines.append(f"    'hmac_secret': {_escape_string(connection.hmac_secret)},")
     lines.append("})")
     lines.append("")
     return "\n".join(lines)

src/tinybird_sdk/migrate/parse_connection.py

@@ -31,6 +31,8 @@
     "S3_ACCESS_KEY",
     "S3_SECRET",
     "GCS_SERVICE_ACCOUNT_CREDENTIALS_JSON",
+    "GCS_HMAC_ACCESS_ID",
+    "GCS_HMAC_SECRET",
     "DYNAMODB_ARN",
     "DYNAMODB_REGION",
 }
@@ -60,6 +62,8 @@ def parse_connection_file(resource: ResourceFile) -> ConnectionModel:
     access_key: str | None = None
     access_secret: str | None = None
     service_account_credentials_json: str | None = None
+    gcs_hmac_access_id: str | None = None
+    gcs_hmac_secret: str | None = None
 
     dynamodb_arn: str | None = None
     dynamodb_region: str | None = None
@@ -125,6 +129,10 @@ def parse_connection_file(resource: ResourceFile) -> ConnectionModel:
             access_secret = parse_quoted_value(value)
         elif name == "GCS_SERVICE_ACCOUNT_CREDENTIALS_JSON":
             service_account_credentials_json = parse_quoted_value(value)
+        elif name == "GCS_HMAC_ACCESS_ID":
+            gcs_hmac_access_id = parse_quoted_value(value)
+        elif name == "GCS_HMAC_SECRET":
+            gcs_hmac_secret = parse_quoted_value(value)
         elif name == "DYNAMODB_ARN":
             dynamodb_arn = parse_quoted_value(value)
         elif name == "DYNAMODB_REGION":
@@ -151,6 +159,8 @@ def parse_connection_file(resource: ResourceFile) -> ConnectionModel:
             or access_key
             or access_secret
             or service_account_credentials_json
+            or gcs_hmac_access_id
+            or gcs_hmac_secret
             or dynamodb_arn
             or dynamodb_region
         ):
@@ -193,6 +203,8 @@ def parse_connection_file(resource: ResourceFile) -> ConnectionModel:
             or schema_registry_url
             or ssl_ca_pem
             or service_account_credentials_json
+            or gcs_hmac_access_id
+            or gcs_hmac_secret
             or dynamodb_arn
             or dynamodb_region
         ):
@@ -261,12 +273,32 @@ def parse_connection_file(resource: ResourceFile) -> ConnectionModel:
                 "Kafka/S3/DynamoDB directives are not valid for gcs connections.",
             )
 
-        if not service_account_credentials_json:
+        has_hmac = bool(gcs_hmac_access_id or gcs_hmac_secret)
+
+        if not service_account_credentials_json and not has_hmac:
+            raise MigrationParseError(
+                resource.file_path,
+                "connection",
+                resource.name,
+                "gcs connections require GCS_SERVICE_ACCOUNT_CREDENTIALS_JSON or both "
+                "GCS_HMAC_ACCESS_ID and GCS_HMAC_SECRET.",
+            )
+
+        if service_account_credentials_json and has_hmac:
+            raise MigrationParseError(
+                resource.file_path,
+                "connection",
+                resource.name,
+                "GCS_SERVICE_ACCOUNT_CREDENTIALS_JSON and GCS_HMAC_ACCESS_ID/GCS_HMAC_SECRET are "
+                "mutually exclusive.",
+            )
+
+        if has_hmac and not (gcs_hmac_access_id and gcs_hmac_secret):
             raise MigrationParseError(
                 resource.file_path,
                 "connection",
                 resource.name,
-                "GCS_SERVICE_ACCOUNT_CREDENTIALS_JSON is required for gcs connections.",
+                "GCS_HMAC_ACCESS_ID and GCS_HMAC_SECRET must be provided together.",
             )
 
         return GCSConnectionModel(
@@ -275,6 +307,8 @@ def parse_connection_file(resource: ResourceFile) -> ConnectionModel:
             file_path=resource.file_path,
             connection_type="gcs",
             service_account_credentials_json=service_account_credentials_json,
+            hmac_access_id=gcs_hmac_access_id,
+            hmac_secret=gcs_hmac_secret,
         )
 
     if connection_type == "dynamodb":
@@ -291,6 +325,8 @@ def parse_connection_file(resource: ResourceFile) -> ConnectionModel:
             or access_key
             or access_secret
             or service_account_credentials_json
+            or gcs_hmac_access_id
+            or gcs_hmac_secret
         ):
             raise MigrationParseError(
                 resource.file_path,

src/tinybird_sdk/migrate/types.py

@@ -216,7 +216,9 @@ class GCSConnectionModel:
     name: str
     file_path: str
     connection_type: Literal["gcs"]
-    service_account_credentials_json: str
+    service_account_credentials_json: str | None = None
+    hmac_access_id: str | None = None
+    hmac_secret: str | None = None
 
 
 @dataclass(frozen=True, slots=True)

src/tinybird_sdk/schema/connection.py

@@ -54,7 +54,9 @@ class S3ConnectionDefinition:
 
 @dataclass(frozen=True, slots=True)
 class GCSConnectionOptions:
-    service_account_credentials_json: str
+    service_account_credentials_json: str | None = None
+    hmac_access_id: str | None = None
+    hmac_secret: str | None = None
 
 
 @dataclass(frozen=True, slots=True)
@@ -126,8 +128,28 @@ def define_gcs_connection(
         options if isinstance(options, GCSConnectionOptions) else GCSConnectionOptions(**options)
     )
 
-    if not normalized.service_account_credentials_json.strip():
-        raise ValueError("GCS connection `service_account_credentials_json` is required.")
+    service_account_credentials_json = normalized.service_account_credentials_json
+    has_service_account = bool(
+        service_account_credentials_json and service_account_credentials_json.strip()
+    )
+    has_hmac = bool(normalized.hmac_access_id or normalized.hmac_secret)
+
+    if not has_service_account and not has_hmac:
+        raise ValueError(
+            "GCS connection requires either `service_account_credentials_json` or both "
+            "`hmac_access_id` and `hmac_secret`."
+        )
+
+    if has_service_account and has_hmac:
+        raise ValueError(
+            "GCS connection `service_account_credentials_json` and `hmac_access_id`/`hmac_secret` "
+            "are mutually exclusive."
+        )
+
+    if has_hmac and not (normalized.hmac_access_id and normalized.hmac_secret):
+        raise ValueError(
+            "GCS connection `hmac_access_id` and `hmac_secret` must be provided together."
+        )
 
     return GCSConnectionDefinition(_name=name, options=normalized)
 

tests/test_gcs_hmac_connection.py

@@ -0,0 +1,158 @@
+from __future__ import annotations
+
+import pytest
+
+import tinybird_sdk as sdk
+from tinybird_sdk import (
+    define_gcs_connection,
+    is_connection_definition,
+    is_gcs_connection_definition,
+)
+from tinybird_sdk.generator.connection import generate_connection
+from tinybird_sdk.migrate.emit_ts import emit_migration_file_content
+from tinybird_sdk.migrate.parse_connection import parse_connection_file
+from tinybird_sdk.migrate.types import ResourceFile
+
+
+def _hmac_connection() -> object:
+    return define_gcs_connection(
+        "landing_gcs_hmac",
+        {
+            "hmac_access_id": "GOOG1EHMACACCESSID",
+            "hmac_secret": "s3cr3t",
+        },
+    )
+
+
+def test_define_gcs_connection_accepts_hmac_key_pair() -> None:
+    connection = _hmac_connection()
+    assert connection._connectionType == "gcs"
+    assert connection._type == "connection"
+    assert is_connection_definition(connection)
+    assert is_gcs_connection_definition(connection)
+    assert sdk.get_connection_type(connection) == "gcs"
+    assert connection.options.hmac_access_id == "GOOG1EHMACACCESSID"
+    assert connection.options.hmac_secret == "s3cr3t"
+    assert connection.options.service_account_credentials_json is None
+
+
+def test_define_gcs_connection_still_accepts_service_account_json() -> None:
+    connection = define_gcs_connection("landing_gcs", {"service_account_credentials_json": "{}"})
+    assert connection.options.service_account_credentials_json == "{}"
+    assert connection.options.hmac_access_id is None
+    assert connection.options.hmac_secret is None
+
+
+def test_define_gcs_connection_requires_an_auth_method() -> None:
+    with pytest.raises(ValueError, match="requires either"):
+        define_gcs_connection("c", {})
+
+
+def test_define_gcs_connection_rejects_mixing_auth_methods() -> None:
+    with pytest.raises(ValueError, match="mutually exclusive"):
+        define_gcs_connection(
+            "c",
+            {
+                "service_account_credentials_json": "{}",
+                "hmac_access_id": "id",
+                "hmac_secret": "secret",
+            },
+        )
+
+
+def test_define_gcs_connection_requires_hmac_pair_together() -> None:
+    with pytest.raises(ValueError, match="must be provided together"):
+        define_gcs_connection("c", {"hmac_access_id": "id"})
+    with pytest.raises(ValueError, match="must be provided together"):
+        define_gcs_connection("c", {"hmac_secret": "secret"})
+
+
+def test_generate_gcs_connection_emits_hmac_directives() -> None:
+    generated = generate_connection(_hmac_connection())
+    assert generated.name == "landing_gcs_hmac"
+    assert generated.content == (
+        "TYPE gcs\nGCS_HMAC_ACCESS_ID GOOG1EHMACACCESSID\nGCS_HMAC_SECRET s3cr3t"
+    )
+
+
+def test_generate_gcs_connection_still_emits_service_account_json() -> None:
+    connection = define_gcs_connection("landing_gcs", {"service_account_credentials_json": "{}"})
+    generated = generate_connection(connection)
+    assert generated.content == "TYPE gcs\nGCS_SERVICE_ACCOUNT_CREDENTIALS_JSON {}"
+
+
+def test_parse_gcs_connection_file_with_hmac_directives() -> None:
+    resource = ResourceFile(
+        kind="connection",
+        file_path="connections/landing_gcs_hmac.connection",
+        absolute_path="/x/connections/landing_gcs_hmac.connection",
+        name="landing_gcs_hmac",
+        content=(
+            "TYPE gcs\nGCS_HMAC_ACCESS_ID GOOG1EHMACACCESSID\nGCS_HMAC_SECRET s3cr3t\n# a comment\n"
+        ),
+    )
+
+    model = parse_connection_file(resource)
+    assert model.connection_type == "gcs"
+    assert model.hmac_access_id == "GOOG1EHMACACCESSID"
+    assert model.hmac_secret == "s3cr3t"
+    assert model.service_account_credentials_json is None
+
+
+def test_parse_gcs_connection_requires_an_auth_method() -> None:
+    base = ResourceFile(
+        kind="connection",
+        file_path="c.connection",
+        absolute_path="/x/c.connection",
+        name="c",
+        content="TYPE gcs\n",
+    )
+    with pytest.raises(Exception, match="require GCS_SERVICE_ACCOUNT_CREDENTIALS_JSON"):
+        parse_connection_file(base)
+
+
+def test_parse_gcs_connection_rejects_mixing_auth_methods() -> None:
+    base = ResourceFile(
+        kind="connection",
+        file_path="c.connection",
+        absolute_path="/x/c.connection",
+        name="c",
+        content=(
+            "TYPE gcs\n"
+            "GCS_SERVICE_ACCOUNT_CREDENTIALS_JSON {}\n"
+            "GCS_HMAC_ACCESS_ID id\n"
+            "GCS_HMAC_SECRET secret\n"
+        ),
+    )
+    with pytest.raises(Exception, match="mutually exclusive"):
+        parse_connection_file(base)
+
+
+def test_parse_gcs_connection_requires_hmac_pair_together() -> None:
+    base = ResourceFile(
+        kind="connection",
+        file_path="c.connection",
+        absolute_path="/x/c.connection",
+        name="c",
+        content="TYPE gcs\nGCS_HMAC_ACCESS_ID id\n",
+    )
+    with pytest.raises(Exception, match="must be provided together"):
+        parse_connection_file(base)
+
+
+def test_emit_ts_round_trip_for_gcs_hmac() -> None:
+    connection_resource = ResourceFile(
+        kind="connection",
+        file_path="connections/landing_gcs_hmac.connection",
+        absolute_path="/x/connections/landing_gcs_hmac.connection",
+        name="landing_gcs_hmac",
+        content=("TYPE gcs\nGCS_HMAC_ACCESS_ID GOOG1EHMACACCESSID\nGCS_HMAC_SECRET s3cr3t\n"),
+    )
+
+    connection = parse_connection_file(connection_resource)
+    output = emit_migration_file_content([connection])
+
+    assert 'landing_gcs_hmac = define_gcs_connection("landing_gcs_hmac"' in output
+    assert "'hmac_access_id': \"GOOG1EHMACACCESSID\"" in output
+    assert "'hmac_secret': \"s3cr3t\"" in output
+    assert "service_account_credentials_json" not in output