feat(cli): register-device --claim mints a device claim and reports its claim_id

CLAUDE.md

@@ -99,6 +99,8 @@ The SDK uses a src layout with the main package at `src/hubblenetwork/`. Public
 
 - **EID modes**: Two EID rotation modes — UNIX_TIME and DEVICE_UPTIME. Device-uptime mode uses a fixed pool size of 128. The `decrypt()` function's `counter_mode` parameter accepts `"UNIX_TIME"` (default) or `"DEVICE_UPTIME"`; CLI commands use `--counter-mode UNIX_TIME|DEVICE_UPTIME`. For AES-128-EAX device registration on DEVICE_UPTIME, the rotation period can be set via `period_seconds` (SDK) / `--period-seconds` (CLI) or `period_exponent` / `--period-exponent` (period = 2^n seconds; cloud accepts 10-15, default 15 ≈ 9h). The two are mutually exclusive.
 
+- **Device claims at registration**: `org register-device --claim` (SDK: `register_device(claim=True)`) sends `device_claims: [{"new_claim_using_device_id": true}]` on the provision request. The provision response never echoes a claim id, but in this mode the cloud names the claim after the device, so `Device.claim_id` is set to the device id client-side. `--claim-destination-org <uuid>` / `claim_destination_org_id` pins the claim's destination and requires `--claim`. Minting needs the `can_create_device_claims` entitlement; the CLI turns the resulting 403 into `Error: entitlement requirements not met` (exit 1) instead of an "Unexpected error". Tests in `tests/test_cli_register_device_claim.py`, `tests/test_register_device_cloud.py`, `tests/test_register_device_validation.py`.
+
 - **Satellite scanning requires Docker**: `sat.scan()` pulls and runs a privileged Docker container. Docker daemon must be running. Raises `DockerError` (not `SatelliteError`) if Docker is unavailable. The `docker` Python package is a required (not optional) dependency.
 
 ### Environment Variables

README.md

@@ -177,6 +177,7 @@ hubblenetwork org get-packets <id> --days 30 --format csv
 hubblenetwork org get-packets <id> -n 50 --debug
 
 hubblenetwork org register-device           # returns the new device's key
+hubblenetwork org register-device --claim   # also mints a claim; prints claim_id
 hubblenetwork org set-device-name <id> <name>
 hubblenetwork org delete-device <id>
 ```
@@ -196,6 +197,11 @@ tabular output when you want the run itself bounded.
 seconds; the cloud accepts 10-15, default 15 ≈ 9h). The two period flags are mutually
 exclusive.
 
+`--claim` also mints a device claim for the new device and prints its `claim_id`,
+which equals the device id. `--claim-destination-org <uuid>` pins the claim's
+destination and requires `--claim`. Minting needs the `can_create_device_claims`
+entitlement; without it the command fails with `entitlement requirements not met`.
+
 
 ## Receive a device's satellite uplink
 
@@ -415,6 +421,7 @@ org = Organization(
 )
 
 new_dev = org.register_device()           # returns a Device, with its key
+claimed = org.register_device(claim=True) # also mints a claim; claimed.claim_id == claimed.id
 for d in org.iter_devices():              # streams as pages arrive
     print(d.id, d.name)
 for pkt in org.iter_packets(new_dev):     # ditto; both take on_page(page, total)

src/hubblenetwork/cli.py

@@ -4170,10 +4170,34 @@ def _format_period_exponent(n: int) -> str:
     show_default=False,
     help="EID rotation period exponent; period = 2^n seconds. Cloud accepts 10-15 (default 15).",
 )
+@click.option(
+    "--claim",
+    is_flag=True,
+    default=False,
+    help="Also mint a device claim for the new device; its claim_id equals the device id.",
+)
+@click.option(
+    "--claim-destination-org",
+    type=str,
+    default=None,
+    show_default=False,
+    metavar="<uuid>",
+    help="Pin the minted claim to this destination organization (requires --claim).",
+)
 @pass_orgcfg
-def register_device(org: Organization, encryption, counter_source, period_seconds, period_exponent) -> None:
+def register_device(
+    org: Organization,
+    encryption,
+    counter_source,
+    period_seconds,
+    period_exponent,
+    claim,
+    claim_destination_org,
+) -> None:
     if period_seconds is not None and period_exponent is not None:
         raise click.UsageError("provide at most one of --period-seconds / --period-exponent")
+    if claim_destination_org is not None and not claim:
+        raise click.UsageError("--claim-destination-org requires --claim")
 
     if encryption:
         click.secho(f'[INFO] Overriding default encryption, using "{encryption}"')
@@ -4194,13 +4218,31 @@ def register_device(org: Organization, encryption, counter_source, period_second
         click.secho(
             f'[INFO] Using default EID rotation period exponent: 15 ({_format_period_exponent(15)})'
         )
+    if claim:
+        destination = f" for organization {claim_destination_org}" if claim_destination_org else ""
+        click.secho(f"[INFO] Minting a device claim{destination}; claim_id will equal the device id")
 
-    click.secho(str(org.register_device(
-        encryption=encryption,
-        counter_source=counter_source,
-        period_seconds=period_seconds,
-        period_exponent=period_exponent,
-    )))
+    try:
+        device = org.register_device(
+            encryption=encryption,
+            counter_source=counter_source,
+            period_seconds=period_seconds,
+            period_exponent=period_exponent,
+            claim=claim,
+            claim_destination_org_id=claim_destination_org,
+        )
+    except BackendError as e:
+        # The cloud rejects claim minting with a 403 when the organization lacks
+        # the entitlement; that is a permissions problem, not an unexpected error.
+        if claim and str(e).startswith("403"):
+            raise click.ClickException(
+                "entitlement requirements not met\n"
+                "  Minting a device claim needs the can_create_device_claims "
+                "entitlement on this organization.\n"
+                "  Ask Hubble to enable it, or register without --claim."
+            ) from e
+        raise
+    click.secho(str(device))
 
 
 @org.command("delete-device", short_help="Delete a device from your organization")

src/hubblenetwork/cloud.py

@@ -174,13 +174,20 @@ def register_device(
     period_in_seconds: int | None = None,
     period_exponent: int | None = None,
     tags: dict[str, str] | None = None,
+    claim: bool = False,
+    claim_destination_org_id: str | None = None,
 ) -> Any:
     """Create a new device and return it.
 
     Args:
         tags: Optional custom tags for the new device. The Cloud API accepts a
             per-device tags map in a list matching ``n_devices``; when provided
             here it is sent as a single-element list.
+        claim: Mint a device claim for the new device. The Cloud API names the
+            claim after the device, so the claim ID equals the device ID; the
+            provision response itself never echoes it back.
+        claim_destination_org_id: Pin the minted claim to a destination
+            organization. Only meaningful with ``claim=True``.
     """
     data: dict = {
         "n_devices": 1,
@@ -195,6 +202,11 @@ def register_device(
         data["eid_rotation"] = eid_rotation
     if tags is not None:
         data["tags"] = [tags]
+    if claim:
+        claim_ref: dict = {"new_claim_using_device_id": True}
+        if claim_destination_org_id is not None:
+            claim_ref["destination_org_id"] = claim_destination_org_id
+        data["device_claims"] = [claim_ref]
     return cloud_request(
         method="POST",
         env=env,

src/hubblenetwork/device.py

@@ -18,6 +18,7 @@ class Device:
     tags: dict[str, str] | None = None
     created_ts: int | None = None
     active: bool | None = False
+    claim_id: str | None = None
 
     def __str__(self) -> str:
         key_str = (
@@ -27,7 +28,8 @@ def __str__(self) -> str:
         )
         return (
             f"Device(id={self.id!r}, key={key_str!r}, name={self.name!r}, "
-            f"tags={self.tags!r}, created_ts={self.created_ts!r}, active={self.active!r})"
+            f"tags={self.tags!r}, created_ts={self.created_ts!r}, active={self.active!r}, "
+            f"claim_id={self.claim_id!r})"
         )
 
     @classmethod

src/hubblenetwork/org.py

@@ -53,10 +53,13 @@ def register_device(
         period_seconds: int | None = None,
         period_exponent: int | None = None,
         tags: dict[str, str] | None = None,
+        claim: bool = False,
+        claim_destination_org_id: str | None = None,
     ) -> Device:
         """
         Register a new device in this organization and return it.
-        Returned Device will have an ID and provisioned key.
+        Returned Device will have an ID and provisioned key, and a claim_id
+        when a claim was minted.
 
         Args:
             encryption: Encryption type ("AES-256-CTR", "AES-128-CTR", "AES-128-EAX", or "NONE").
@@ -68,7 +71,13 @@ def register_device(
                              when encryption='AES-128-EAX' and counter_source='DEVICE_UPTIME'.
                              Mutually exclusive with period_seconds.
             tags: Optional custom key/value tags applied at registration.
+            claim: Mint a device claim for the new device (claim_id equals the
+                   device id). Requires the can_create_device_claims entitlement.
+            claim_destination_org_id: Pin the minted claim to a destination
+                                      organization. Requires claim=True.
         """
+        if claim_destination_org_id is not None and not claim:
+            raise ValidationError("claim_destination_org_id requires claim=True")
         if counter_source is not None and counter_source not in _VALID_COUNTER_SOURCES:
             raise ValidationError(
                 f"counter_source must be one of {sorted(_VALID_COUNTER_SOURCES)}, got {counter_source!r}"
@@ -94,10 +103,17 @@ def register_device(
             period_in_seconds=period_seconds,
             period_exponent=period_exponent,
             tags=tags,
+            claim=claim,
+            claim_destination_org_id=claim_destination_org_id,
         )
         device = resp["devices"][0]
         key_bytes = base64.b64decode(device["key"]) if device.get("key") else None
-        return Device(id=device["device_id"], key=key_bytes)
+        device_id = device["device_id"]
+        return Device(
+            id=device_id,
+            key=key_bytes,
+            claim_id=device_id if claim else None,
+        )
 
     def set_device_name(self, device_id: str, name: str) -> Device:
         """

tests/test_cli_register_device_claim.py

@@ -0,0 +1,64 @@
+from unittest.mock import MagicMock, patch
+
+from click.testing import CliRunner
+
+from hubblenetwork.cli import cli
+from hubblenetwork.device import Device
+
+ORG = "11111111-2222-3333-4444-555555555555"
+
+
+def _run(args):
+    org = MagicMock()
+    org.register_device.return_value = Device(id="d1", key=b"abc", claim_id="d1")
+    with patch("hubblenetwork.cli.Organization", return_value=org):
+        result = CliRunner().invoke(
+            cli, ["org", "register-device", *args],
+            env={"HUBBLE_ORG_ID": "o", "HUBBLE_API_TOKEN": "t"},
+        )
+    return result, org
+
+
+class TestRegisterDeviceClaimOption:
+    def test_claim_flag_forwarded_and_claim_id_printed(self):
+        result, org = _run(["--claim"])
+        assert result.exit_code == 0, result.output + str(result.exception)
+        kwargs = org.register_device.call_args.kwargs
+        assert kwargs["claim"] is True
+        assert kwargs["claim_destination_org_id"] is None
+        assert "claim_id='d1'" in result.stdout
+
+    def test_claim_destination_org_forwarded(self):
+        result, org = _run(["--claim", "--claim-destination-org", ORG])
+        assert result.exit_code == 0, result.output + str(result.exception)
+        assert org.register_device.call_args.kwargs["claim_destination_org_id"] == ORG
+
+    def test_destination_without_claim_is_usage_error(self):
+        result, org = _run(["--claim-destination-org", ORG])
+        assert result.exit_code == 2
+        assert "--claim-destination-org requires --claim" in result.stderr
+        org.register_device.assert_not_called()
+
+    def test_default_does_not_claim(self):
+        result, org = _run([])
+        assert result.exit_code == 0, result.output + str(result.exception)
+        assert org.register_device.call_args.kwargs["claim"] is False
+
+
+class TestClaimEntitlementError:
+    def test_403_on_claim_reports_entitlement_requirements_not_met(self):
+        from hubblenetwork.errors import BackendError
+
+        org = MagicMock()
+        org.register_device.side_effect = BackendError(
+            "403: originating a device claim requires the can_create_device_claims "
+            "entitlement for this organization"
+        )
+        with patch("hubblenetwork.cli.Organization", return_value=org):
+            result = CliRunner().invoke(
+                cli, ["org", "register-device", "--claim"],
+                env={"HUBBLE_ORG_ID": "o", "HUBBLE_API_TOKEN": "t"},
+            )
+        assert result.exit_code == 1
+        assert "entitlement requirements not met" in result.output
+        assert "can_create_device_claims" in result.output

tests/test_register_device_cloud.py

@@ -148,3 +148,35 @@ def test_explicit_tags_included(self, mock_request, credentials, env):
             "set_name": "named",
             "set_tags": {"satellite": "next-pass"},
         }
+
+
+class TestRegisterDeviceClaimBody:
+    @patch("hubblenetwork.cloud.cloud_request")
+    def test_claim_mints_new_claim_using_device_id(self, mock_request, credentials, env):
+        mock_request.return_value = ({"devices": [{"device_id": "d1", "key": "abc="}]}, None)
+        register_device(credentials=credentials, env=env, claim=True)
+        body = mock_request.call_args.kwargs["json"]
+        assert body["device_claims"] == [{"new_claim_using_device_id": True}]
+
+    @patch("hubblenetwork.cloud.cloud_request")
+    def test_claim_destination_org_pins_destination(self, mock_request, credentials, env):
+        mock_request.return_value = ({"devices": [{"device_id": "d1", "key": "abc="}]}, None)
+        register_device(
+            credentials=credentials,
+            env=env,
+            claim=True,
+            claim_destination_org_id="11111111-2222-3333-4444-555555555555",
+        )
+        body = mock_request.call_args.kwargs["json"]
+        assert body["device_claims"] == [
+            {
+                "new_claim_using_device_id": True,
+                "destination_org_id": "11111111-2222-3333-4444-555555555555",
+            }
+        ]
+
+    @patch("hubblenetwork.cloud.cloud_request")
+    def test_device_claims_omitted_by_default(self, mock_request, credentials, env):
+        mock_request.return_value = ({"devices": [{"device_id": "d1", "key": "abc="}]}, None)
+        register_device(credentials=credentials, env=env)
+        assert "device_claims" not in mock_request.call_args.kwargs["json"]

tests/test_register_device_validation.py

@@ -101,3 +101,35 @@ def test_tags_default_none(self, mock_reg, org):
         org.register_device()
         kwargs = mock_reg.call_args.kwargs
         assert kwargs["tags"] is None
+
+
+class TestClaim:
+    def test_destination_requires_claim(self, org):
+        with pytest.raises(ValidationError, match="claim_destination_org_id requires claim=True"):
+            org.register_device(claim_destination_org_id="11111111-2222-3333-4444-555555555555")
+
+    @patch("hubblenetwork.org.cloud.register_device")
+    def test_claim_forwarded(self, mock_reg, org):
+        mock_reg.return_value = {"devices": [{"device_id": "d1", "key": "YWJj"}]}
+        org.register_device(claim=True, claim_destination_org_id="11111111-2222-3333-4444-555555555555")
+        kwargs = mock_reg.call_args.kwargs
+        assert kwargs["claim"] is True
+        assert kwargs["claim_destination_org_id"] == "11111111-2222-3333-4444-555555555555"
+
+    @patch("hubblenetwork.org.cloud.register_device")
+    def test_claim_id_equals_device_id_when_claimed(self, mock_reg, org):
+        mock_reg.return_value = {"devices": [{"device_id": "d1", "key": "YWJj"}]}
+        device = org.register_device(claim=True)
+        assert device.claim_id == "d1"
+        assert "claim_id='d1'" in str(device)
+
+    @patch("hubblenetwork.org.cloud.register_device")
+    def test_claim_id_none_when_not_claimed(self, mock_reg, org):
+        mock_reg.return_value = {"devices": [{"device_id": "d1", "key": "YWJj"}]}
+        device = org.register_device()
+        assert device.claim_id is None
+        assert kwargs_claim(mock_reg) is False
+
+
+def kwargs_claim(mock_reg):
+    return mock_reg.call_args.kwargs["claim"]