Skip to content

Hub clients

xcore_agent.agent.hub_client — the HubClient protocol, the in-memory test double, and the HttpHubClient implementation of the proposed .xdeploy REST contract.

xcore_agent.agent.hub_client

Client interface for talking to XCore Hub.

HubClient is the contract the deployment pipeline depends on. HttpHubClient implements it against a REST contract proposed here — inferred from the project's architecture notes (XDevKey vs. deployment credential, the authenticate -> request_artifact -> download -> obtain_deployment_key -> notify sequence, the DeploymentReport shape) rather than from a published Hub API spec, because XCore Hub itself does not exist yet. Treat the routes below as a concrete starting point for building the real Hub against, or adjust this client once the real Hub's endpoints are decided — nothing else in xcore-agent needs to change either way, since everything else talks to HubClient the protocol, not to HttpHubClient the implementation.

Proposed contract (all bodies JSON, bearer auth via the session's access token except /v1/auth itself; binary fields are base64):

POST /v1/auth
    -> {xdevkey, project_id}
    <- {access_token}

GET /v1/projects/{project_id}/versions/latest
    <- {version}

GET /v1/projects/{project_id}/artifacts/{version}
    <- {download_url, signature, signer_public_key}

GET <download_url>                          (may be a different host,
    <- raw bytes                              e.g. signed blob storage)

POST /v1/deployments/authorize
    -> {deployment_credential, artifact_signature}
    <- {dek}                                  (revocation is enforced here)

POST /v1/deployments/report
    -> DeploymentReport fields
    <- {deployment_id}

POST /v1/projects/{project_id}/publish                      (validated against
    -> multipart/form-data:                                   a real XCore Hub —
         version, project_name, content_sha256,                see app/xdeploy in
         dek (base64), signature (base64),                     the Marketplace repo)
         signer_public_key (base64), artifact (file)
    -> header: X-API-Key: <xdevkey>       (not session-bearer — publish is a
                                             local build-time act, not a
                                             deployment; no prior /v1/auth needed)
    <- {artifact_id, project_id, version, content_sha256, size_bytes, created_at}

Session dataclass

An authenticated session against XCore Hub.

Source code in xcore_agent/agent/hub_client.py 
59
60
61
62
63
64
@dataclass(frozen=True)
class Session:
    """An authenticated session against XCore Hub."""

    project_id: str
    access_token: str
project_id: str instance-attribute
access_token: str instance-attribute
__init__(project_id: str, access_token: str) -> None

ArtifactLocation dataclass

Where to fetch a specific artifact version from, plus the outer signature so the agent can verify authenticity before spending time decrypting a (possibly large) ciphertext body.

Source code in xcore_agent/agent/hub_client.py 
67
68
69
70
71
72
73
74
75
@dataclass(frozen=True)
class ArtifactLocation:
    """Where to fetch a specific artifact version from, plus the outer
    signature so the agent can verify authenticity before spending time
    decrypting a (possibly large) ciphertext body."""

    download_url: str
    signature: bytes
    signer_public_key: bytes
download_url: str instance-attribute
signature: bytes instance-attribute
signer_public_key: bytes instance-attribute
__init__(download_url: str, signature: bytes, signer_public_key: bytes) -> None

PublishResult dataclass

What XCore Hub confirms after accepting a newly published artifact.

Source code in xcore_agent/agent/hub_client.py 
78
79
80
81
82
83
84
85
86
87
@dataclass(frozen=True)
class PublishResult:
    """What XCore Hub confirms after accepting a newly published artifact."""

    artifact_id: str
    project_id: str
    version: str
    content_sha256: str
    size_bytes: int
    created_at: str
artifact_id: str instance-attribute
project_id: str instance-attribute
version: str instance-attribute
content_sha256: str instance-attribute
size_bytes: int instance-attribute
created_at: str instance-attribute
__init__(artifact_id: str, project_id: str, version: str, content_sha256: str, size_bytes: int, created_at: str) -> None

DeploymentReport dataclass

Source code in xcore_agent/agent/hub_client.py 
90
91
92
93
94
95
96
97
98
@dataclass(frozen=True)
class DeploymentReport:
    project_id: str
    deployment_id: str
    status: str  # "success" | "failed"
    version: str
    started_at: str
    completed_at: str
    plugins: list[dict]
project_id: str instance-attribute
deployment_id: str instance-attribute
status: str instance-attribute
version: str instance-attribute
started_at: str instance-attribute
completed_at: str instance-attribute
plugins: list[dict] instance-attribute
__init__(project_id: str, deployment_id: str, status: str, version: str, started_at: str, completed_at: str, plugins: list[dict]) -> None

HubClient

Bases: Protocol

Source code in xcore_agent/agent/hub_client.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
class HubClient(Protocol):
    async def authenticate(self, *, xdevkey: str, project_id: str) -> Session: ...

    async def get_latest_version(self, session: Session, *, project_id: str) -> str:
        """Return the version/tag XCore Hub currently considers the latest
        release for this project. This is the primitive the CI/CD watch loop
        (`agent.watcher.Watcher`) polls to notice a new version has been
        published and trigger an automatic redeploy."""
        ...

    async def request_artifact(self, session: Session, *, version: str) -> ArtifactLocation: ...

    async def download(self, location: ArtifactLocation) -> bytes: ...

    async def obtain_deployment_key(
        self, session: Session, *, deployment_credential: str, artifact_signature: bytes
    ) -> bytes:
        """Ask the Hub to unwrap and return the artifact's DEK.

        The Hub is the only party that ever holds the KEK; this call is also
        where access revocation is enforced (a revoked XDevKey or expired
        deployment credential must be rejected here, before any bytes are
        decrypted) — see the project's architecture notes.
        """
        ...

    async def notify(self, session: Session, report: DeploymentReport) -> None: ...

    async def publish(
        self,
        *,
        xdevkey: str,
        project_id: str,
        project_name: str,
        version: str,
        ciphertext: bytes,
        content_sha256: str,
        dek: bytes,
        signature: bytes,
        signer_public_key: bytes,
    ) -> PublishResult:
        """Upload a freshly built artifact (see `packer.builder.build_artifact`)
        to XCore Hub for storage. Authenticated by raw `xdevkey`, not a
        `Session` — publishing is a local build-time act performed before
        any agent ever calls `authenticate`/`/v1/auth` for this artifact."""
        ...
authenticate(*, xdevkey: str, project_id: str) -> Session async
Source code in xcore_agent/agent/hub_client.py 
102
async def authenticate(self, *, xdevkey: str, project_id: str) -> Session: ...
get_latest_version(session: Session, *, project_id: str) -> str async

Return the version/tag XCore Hub currently considers the latest release for this project. This is the primitive the CI/CD watch loop (agent.watcher.Watcher) polls to notice a new version has been published and trigger an automatic redeploy.

Source code in xcore_agent/agent/hub_client.py 
104
105
106
107
108
109
async def get_latest_version(self, session: Session, *, project_id: str) -> str:
    """Return the version/tag XCore Hub currently considers the latest
    release for this project. This is the primitive the CI/CD watch loop
    (`agent.watcher.Watcher`) polls to notice a new version has been
    published and trigger an automatic redeploy."""
    ...
request_artifact(session: Session, *, version: str) -> ArtifactLocation async
Source code in xcore_agent/agent/hub_client.py 
111
async def request_artifact(self, session: Session, *, version: str) -> ArtifactLocation: ...
download(location: ArtifactLocation) -> bytes async
Source code in xcore_agent/agent/hub_client.py 
113
async def download(self, location: ArtifactLocation) -> bytes: ...
obtain_deployment_key(session: Session, *, deployment_credential: str, artifact_signature: bytes) -> bytes async

Ask the Hub to unwrap and return the artifact's DEK.

The Hub is the only party that ever holds the KEK; this call is also where access revocation is enforced (a revoked XDevKey or expired deployment credential must be rejected here, before any bytes are decrypted) — see the project's architecture notes.

Source code in xcore_agent/agent/hub_client.py 
115
116
117
118
119
120
121
122
123
124
125
async def obtain_deployment_key(
    self, session: Session, *, deployment_credential: str, artifact_signature: bytes
) -> bytes:
    """Ask the Hub to unwrap and return the artifact's DEK.

    The Hub is the only party that ever holds the KEK; this call is also
    where access revocation is enforced (a revoked XDevKey or expired
    deployment credential must be rejected here, before any bytes are
    decrypted) — see the project's architecture notes.
    """
    ...
notify(session: Session, report: DeploymentReport) -> None async
Source code in xcore_agent/agent/hub_client.py 
127
async def notify(self, session: Session, report: DeploymentReport) -> None: ...
publish(*, xdevkey: str, project_id: str, project_name: str, version: str, ciphertext: bytes, content_sha256: str, dek: bytes, signature: bytes, signer_public_key: bytes) -> PublishResult async

Upload a freshly built artifact (see packer.builder.build_artifact) to XCore Hub for storage. Authenticated by raw xdevkey, not a Session — publishing is a local build-time act performed before any agent ever calls authenticate//v1/auth for this artifact.

Source code in xcore_agent/agent/hub_client.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
async def publish(
    self,
    *,
    xdevkey: str,
    project_id: str,
    project_name: str,
    version: str,
    ciphertext: bytes,
    content_sha256: str,
    dek: bytes,
    signature: bytes,
    signer_public_key: bytes,
) -> PublishResult:
    """Upload a freshly built artifact (see `packer.builder.build_artifact`)
    to XCore Hub for storage. Authenticated by raw `xdevkey`, not a
    `Session` — publishing is a local build-time act performed before
    any agent ever calls `authenticate`/`/v1/auth` for this artifact."""
    ...

InMemoryHubClient dataclass

HubClient implemented entirely in memory, serving one pre-built artifact — for exercising DeploymentRunner end-to-end (signature verification, AES-GCM decryption, manifest/content-hash checks, install/rollback) without a live Hub, since HttpHubClient speaks to a Hub that does not exist yet (see this module's docstring).

Not a test-only mock kept private to one test file: it is the supported way to drive the real pipeline against a real, locally-built .xdeploy artifact — construct it with the exact (ciphertext, dek, signature, signer_public_key) tuple packer.builder.build_artifact/seal_directory returned, and it will behave, from DeploymentRunner's point of view, like a Hub that already has that artifact stored and ready to serve. notified records every notify() call for the caller to assert on.

Source code in xcore_agent/agent/hub_client.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
@dataclass
class InMemoryHubClient:
    """`HubClient` implemented entirely in memory, serving one pre-built
    artifact — for exercising `DeploymentRunner` end-to-end (signature
    verification, AES-GCM decryption, manifest/content-hash checks,
    install/rollback) without a live Hub, since `HttpHubClient` speaks to a
    Hub that does not exist yet (see this module's docstring).

    Not a test-only mock kept private to one test file: it is the supported
    way to drive the real pipeline against a real, locally-built `.xdeploy`
    artifact — construct it with the exact `(ciphertext, dek, signature,
    signer_public_key)` tuple `packer.builder.build_artifact`/`seal_directory`
    returned, and it will behave, from `DeploymentRunner`'s point of view,
    like a Hub that already has that artifact stored and ready to serve.
    `notified` records every `notify()` call for the caller to assert on.
    """

    ciphertext: bytes
    dek: bytes
    signature: bytes
    signer_public_key: bytes
    notified: list[DeploymentReport] = None  # type: ignore[assignment]

    def __post_init__(self) -> None:
        if self.notified is None:
            self.notified = []

    async def authenticate(self, *, xdevkey: str, project_id: str) -> Session:
        return Session(project_id=project_id, access_token="in-memory-session")

    async def get_latest_version(self, session: Session, *, project_id: str) -> str:
        raise NotImplementedError(
            "InMemoryHubClient serves one fixed artifact — it has no notion of "
            "'latest version' distinct from what it was constructed with; the "
            "watch loop (agent.watcher.Watcher) has no use for this client"
        )

    async def request_artifact(self, session: Session, *, version: str) -> ArtifactLocation:
        return ArtifactLocation(
            download_url="in-memory://artifact",
            signature=self.signature,
            signer_public_key=self.signer_public_key,
        )

    async def download(self, location: ArtifactLocation) -> bytes:
        return self.ciphertext

    async def obtain_deployment_key(
        self, session: Session, *, deployment_credential: str, artifact_signature: bytes
    ) -> bytes:
        return self.dek

    async def notify(self, session: Session, report: DeploymentReport) -> None:
        self.notified.append(report)

    async def publish(
        self,
        *,
        xdevkey: str,
        project_id: str,
        project_name: str,
        version: str,
        ciphertext: bytes,
        content_sha256: str,
        dek: bytes,
        signature: bytes,
        signer_public_key: bytes,
    ) -> PublishResult:
        raise NotImplementedError(
            "InMemoryHubClient is pre-seeded with one fixed artifact for "
            "exercising DeploymentRunner — it has no publish-time storage to "
            "accept a new one into; nothing in the deployment pipeline calls "
            "publish() anyway (it's a CLI build-time act, see cli.py::publish)"
        )
ciphertext: bytes instance-attribute
dek: bytes instance-attribute
signature: bytes instance-attribute
signer_public_key: bytes instance-attribute
notified: list[DeploymentReport] = None class-attribute instance-attribute
__init__(ciphertext: bytes, dek: bytes, signature: bytes, signer_public_key: bytes, notified: list[DeploymentReport] = None) -> None
__post_init__() -> None
Source code in xcore_agent/agent/hub_client.py 
172
173
174
def __post_init__(self) -> None:
    if self.notified is None:
        self.notified = []
authenticate(*, xdevkey: str, project_id: str) -> Session async
Source code in xcore_agent/agent/hub_client.py 
176
177
async def authenticate(self, *, xdevkey: str, project_id: str) -> Session:
    return Session(project_id=project_id, access_token="in-memory-session")
get_latest_version(session: Session, *, project_id: str) -> str async
Source code in xcore_agent/agent/hub_client.py 
179
180
181
182
183
184
async def get_latest_version(self, session: Session, *, project_id: str) -> str:
    raise NotImplementedError(
        "InMemoryHubClient serves one fixed artifact — it has no notion of "
        "'latest version' distinct from what it was constructed with; the "
        "watch loop (agent.watcher.Watcher) has no use for this client"
    )
request_artifact(session: Session, *, version: str) -> ArtifactLocation async
Source code in xcore_agent/agent/hub_client.py 
186
187
188
189
190
191
async def request_artifact(self, session: Session, *, version: str) -> ArtifactLocation:
    return ArtifactLocation(
        download_url="in-memory://artifact",
        signature=self.signature,
        signer_public_key=self.signer_public_key,
    )
download(location: ArtifactLocation) -> bytes async
Source code in xcore_agent/agent/hub_client.py 
193
194
async def download(self, location: ArtifactLocation) -> bytes:
    return self.ciphertext
obtain_deployment_key(session: Session, *, deployment_credential: str, artifact_signature: bytes) -> bytes async
Source code in xcore_agent/agent/hub_client.py 
196
197
198
199
async def obtain_deployment_key(
    self, session: Session, *, deployment_credential: str, artifact_signature: bytes
) -> bytes:
    return self.dek
notify(session: Session, report: DeploymentReport) -> None async
Source code in xcore_agent/agent/hub_client.py 
201
202
async def notify(self, session: Session, report: DeploymentReport) -> None:
    self.notified.append(report)
publish(*, xdevkey: str, project_id: str, project_name: str, version: str, ciphertext: bytes, content_sha256: str, dek: bytes, signature: bytes, signer_public_key: bytes) -> PublishResult async
Source code in xcore_agent/agent/hub_client.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
async def publish(
    self,
    *,
    xdevkey: str,
    project_id: str,
    project_name: str,
    version: str,
    ciphertext: bytes,
    content_sha256: str,
    dek: bytes,
    signature: bytes,
    signer_public_key: bytes,
) -> PublishResult:
    raise NotImplementedError(
        "InMemoryHubClient is pre-seeded with one fixed artifact for "
        "exercising DeploymentRunner — it has no publish-time storage to "
        "accept a new one into; nothing in the deployment pipeline calls "
        "publish() anyway (it's a CLI build-time act, see cli.py::publish)"
    )

HttpHubClient

HTTP implementation of HubClient against the real xcore-team/xdeploy plugin (app/xdeploy) — validated end to end against a live Hub (build -> publish -> a real POST /app/xdeploy/v1/projects/{id}/publish that stored the artifact).

base_url is the Hub's bare root (no /app/... segment — same convention as marketplace_client.MarketplaceClient); every request below prepends its own /app/xdeploy mount internally so callers never need to know this backend's internal plugin layout. Every xcore plugin (including this one) is mounted under /app/<plugin-name> by the xcore router — the previous bare /v1/... paths here were a "proposed contract" written before any real Hub existed to validate against, and were 404ing against the real one.

transport exists so tests can inject httpx.MockTransport instead of hitting the network; production callers leave it as None.

Source code in xcore_agent/agent/hub_client.py 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
class HttpHubClient:
    """HTTP implementation of `HubClient` against the real xcore-team/xdeploy
    plugin (`app/xdeploy`) — validated end to end against a live Hub
    (build -> publish -> a real `POST /app/xdeploy/v1/projects/{id}/publish`
    that stored the artifact).

    `base_url` is the Hub's bare root (no `/app/...` segment — same
    convention as `marketplace_client.MarketplaceClient`); every request
    below prepends its own `/app/xdeploy` mount internally so callers never
    need to know this backend's internal plugin layout. Every xcore plugin
    (including this one) is mounted under `/app/<plugin-name>` by the xcore
    router — the previous bare `/v1/...` paths here were a "proposed
    contract" written before any real Hub existed to validate against, and
    were 404ing against the real one.

    `transport` exists so tests can inject `httpx.MockTransport` instead of
    hitting the network; production callers leave it as `None`.
    """

    def __init__(
        self,
        base_url: str,
        *,
        timeout: float = 30.0,
        transport: httpx.AsyncBaseTransport | None = None,
    ) -> None:
        self._client = httpx.AsyncClient(
            base_url=base_url.rstrip("/"),
            timeout=timeout,
            transport=transport,
            # `download_url` (from request_artifact) has come back as a bare
            # `http://` URL in production while the Hub itself is HTTPS-only
            # — the reverse proxy 301s to https://, and httpx does NOT
            # follow redirects by default. Without this, `download()` was
            # silently returning the redirect response's ~17-byte HTML body
            # ("Moved Permanently") instead of the artifact, which then
            # failed signature verification with no indication that the
            # actual problem was a followed-nowhere redirect, not a bad
            # signature. Real artifacts should never redirect again after
            # this, but there is no security cost to allowing it either:
            # every response we actually trust is authenticated by content
            # (Ed25519 signature over the downloaded bytes, HMAC for the
            # marketplace) rather than by transport, so a redirect changes
            # where the bytes came from, never whether they're trusted.
            follow_redirects=True,
        )

    async def __aenter__(self) -> "HttpHubClient":
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> None:
        await self.aclose()

    async def aclose(self) -> None:
        await self._client.aclose()

    async def authenticate(self, *, xdevkey: str, project_id: str) -> Session:
        response = await self._client.post(
            f"{_MOUNT}/v1/auth", json={"xdevkey": xdevkey, "project_id": project_id}
        )
        if response.status_code in (401, 403):
            raise AuthenticationError(f"authentication failed: {_error_message(response)}")
        _raise_for_status(response, DeploymentError, "authenticate")
        return Session(project_id=project_id, access_token=response.json()["access_token"])

    async def get_latest_version(self, session: Session, *, project_id: str) -> str:
        response = await self._client.get(
            f"{_MOUNT}/v1/projects/{project_id}/versions/latest", headers=_auth_header(session)
        )
        _raise_for_status(response, ArtifactError, "get_latest_version")
        return str(response.json()["version"])

    async def request_artifact(self, session: Session, *, version: str) -> ArtifactLocation:
        response = await self._client.get(
            f"{_MOUNT}/v1/projects/{session.project_id}/artifacts/{version}",
            headers=_auth_header(session),
        )
        _raise_for_status(response, ArtifactError, "request_artifact")
        data = response.json()
        return ArtifactLocation(
            download_url=data["download_url"],
            signature=base64.b64decode(data["signature"]),
            signer_public_key=base64.b64decode(data["signer_public_key"]),
        )

    async def download(self, location: ArtifactLocation) -> bytes:
        response = await self._client.get(location.download_url)
        _raise_for_status(response, ArtifactError, "download")
        return response.content

    async def obtain_deployment_key(
        self, session: Session, *, deployment_credential: str, artifact_signature: bytes
    ) -> bytes:
        response = await self._client.post(
            f"{_MOUNT}/v1/deployments/authorize",
            headers=_auth_header(session),
            json={
                "deployment_credential": deployment_credential,
                "artifact_signature": base64.b64encode(artifact_signature).decode("ascii"),
            },
        )
        if response.status_code in (401, 403):
            raise AuthenticationError(
                f"deployment key authorization denied: {_error_message(response)}"
            )
        _raise_for_status(response, ArtifactError, "obtain_deployment_key")
        return base64.b64decode(response.json()["dek"])

    async def notify(self, session: Session, report: DeploymentReport) -> None:
        response = await self._client.post(
            f"{_MOUNT}/v1/deployments/report",
            headers=_auth_header(session),
            json={
                "project_id": report.project_id,
                "deployment_id": report.deployment_id,
                "status": report.status,
                "version": report.version,
                "started_at": report.started_at,
                "completed_at": report.completed_at,
                "plugins": report.plugins,
            },
        )
        _raise_for_status(response, DeploymentError, "notify")

    async def publish(
        self,
        *,
        xdevkey: str,
        project_id: str,
        project_name: str,
        version: str,
        ciphertext: bytes,
        content_sha256: str,
        dek: bytes,
        signature: bytes,
        signer_public_key: bytes,
    ) -> PublishResult:
        response = await self._client.post(
            f"{_MOUNT}/v1/projects/{project_id}/publish",
            headers={"X-API-Key": xdevkey},
            data={
                "version": version,
                "project_name": project_name,
                "content_sha256": content_sha256,
                "dek": base64.b64encode(dek).decode("ascii"),
                "signature": base64.b64encode(signature).decode("ascii"),
                "signer_public_key": base64.b64encode(signer_public_key).decode("ascii"),
            },
            files={"artifact": ("artifact.xdeploy", ciphertext, "application/octet-stream")},
        )
        if response.status_code in (401, 403):
            raise AuthenticationError(f"publish denied: {_error_message(response)}")
        _raise_for_status(response, PublishError, "publish")
        data = response.json()
        return PublishResult(
            artifact_id=data["artifact_id"],
            project_id=data["project_id"],
            version=data["version"],
            content_sha256=data["content_sha256"],
            size_bytes=data["size_bytes"],
            created_at=data["created_at"],
        )
__init__(base_url: str, *, timeout: float = 30.0, transport: httpx.AsyncBaseTransport | None = None) -> None
Source code in xcore_agent/agent/hub_client.py 
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
def __init__(
    self,
    base_url: str,
    *,
    timeout: float = 30.0,
    transport: httpx.AsyncBaseTransport | None = None,
) -> None:
    self._client = httpx.AsyncClient(
        base_url=base_url.rstrip("/"),
        timeout=timeout,
        transport=transport,
        # `download_url` (from request_artifact) has come back as a bare
        # `http://` URL in production while the Hub itself is HTTPS-only
        # — the reverse proxy 301s to https://, and httpx does NOT
        # follow redirects by default. Without this, `download()` was
        # silently returning the redirect response's ~17-byte HTML body
        # ("Moved Permanently") instead of the artifact, which then
        # failed signature verification with no indication that the
        # actual problem was a followed-nowhere redirect, not a bad
        # signature. Real artifacts should never redirect again after
        # this, but there is no security cost to allowing it either:
        # every response we actually trust is authenticated by content
        # (Ed25519 signature over the downloaded bytes, HMAC for the
        # marketplace) rather than by transport, so a redirect changes
        # where the bytes came from, never whether they're trusted.
        follow_redirects=True,
    )
__aenter__() -> HttpHubClient async
Source code in xcore_agent/agent/hub_client.py 
296
297
async def __aenter__(self) -> "HttpHubClient":
    return self
__aexit__(exc_type: type[BaseException] | None, exc: BaseException | None, tb: TracebackType | None) -> None async
Source code in xcore_agent/agent/hub_client.py 
299
300
301
302
303
304
305
async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc: BaseException | None,
    tb: TracebackType | None,
) -> None:
    await self.aclose()
aclose() -> None async
Source code in xcore_agent/agent/hub_client.py 
307
308
async def aclose(self) -> None:
    await self._client.aclose()
authenticate(*, xdevkey: str, project_id: str) -> Session async
Source code in xcore_agent/agent/hub_client.py 
310
311
312
313
314
315
316
317
async def authenticate(self, *, xdevkey: str, project_id: str) -> Session:
    response = await self._client.post(
        f"{_MOUNT}/v1/auth", json={"xdevkey": xdevkey, "project_id": project_id}
    )
    if response.status_code in (401, 403):
        raise AuthenticationError(f"authentication failed: {_error_message(response)}")
    _raise_for_status(response, DeploymentError, "authenticate")
    return Session(project_id=project_id, access_token=response.json()["access_token"])
get_latest_version(session: Session, *, project_id: str) -> str async
Source code in xcore_agent/agent/hub_client.py 
319
320
321
322
323
324
async def get_latest_version(self, session: Session, *, project_id: str) -> str:
    response = await self._client.get(
        f"{_MOUNT}/v1/projects/{project_id}/versions/latest", headers=_auth_header(session)
    )
    _raise_for_status(response, ArtifactError, "get_latest_version")
    return str(response.json()["version"])
request_artifact(session: Session, *, version: str) -> ArtifactLocation async
Source code in xcore_agent/agent/hub_client.py 
326
327
328
329
330
331
332
333
334
335
336
337
async def request_artifact(self, session: Session, *, version: str) -> ArtifactLocation:
    response = await self._client.get(
        f"{_MOUNT}/v1/projects/{session.project_id}/artifacts/{version}",
        headers=_auth_header(session),
    )
    _raise_for_status(response, ArtifactError, "request_artifact")
    data = response.json()
    return ArtifactLocation(
        download_url=data["download_url"],
        signature=base64.b64decode(data["signature"]),
        signer_public_key=base64.b64decode(data["signer_public_key"]),
    )
download(location: ArtifactLocation) -> bytes async
Source code in xcore_agent/agent/hub_client.py 
339
340
341
342
async def download(self, location: ArtifactLocation) -> bytes:
    response = await self._client.get(location.download_url)
    _raise_for_status(response, ArtifactError, "download")
    return response.content
obtain_deployment_key(session: Session, *, deployment_credential: str, artifact_signature: bytes) -> bytes async
Source code in xcore_agent/agent/hub_client.py 
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
async def obtain_deployment_key(
    self, session: Session, *, deployment_credential: str, artifact_signature: bytes
) -> bytes:
    response = await self._client.post(
        f"{_MOUNT}/v1/deployments/authorize",
        headers=_auth_header(session),
        json={
            "deployment_credential": deployment_credential,
            "artifact_signature": base64.b64encode(artifact_signature).decode("ascii"),
        },
    )
    if response.status_code in (401, 403):
        raise AuthenticationError(
            f"deployment key authorization denied: {_error_message(response)}"
        )
    _raise_for_status(response, ArtifactError, "obtain_deployment_key")
    return base64.b64decode(response.json()["dek"])
notify(session: Session, report: DeploymentReport) -> None async
Source code in xcore_agent/agent/hub_client.py 
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
async def notify(self, session: Session, report: DeploymentReport) -> None:
    response = await self._client.post(
        f"{_MOUNT}/v1/deployments/report",
        headers=_auth_header(session),
        json={
            "project_id": report.project_id,
            "deployment_id": report.deployment_id,
            "status": report.status,
            "version": report.version,
            "started_at": report.started_at,
            "completed_at": report.completed_at,
            "plugins": report.plugins,
        },
    )
    _raise_for_status(response, DeploymentError, "notify")
publish(*, xdevkey: str, project_id: str, project_name: str, version: str, ciphertext: bytes, content_sha256: str, dek: bytes, signature: bytes, signer_public_key: bytes) -> PublishResult async
Source code in xcore_agent/agent/hub_client.py 
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
async def publish(
    self,
    *,
    xdevkey: str,
    project_id: str,
    project_name: str,
    version: str,
    ciphertext: bytes,
    content_sha256: str,
    dek: bytes,
    signature: bytes,
    signer_public_key: bytes,
) -> PublishResult:
    response = await self._client.post(
        f"{_MOUNT}/v1/projects/{project_id}/publish",
        headers={"X-API-Key": xdevkey},
        data={
            "version": version,
            "project_name": project_name,
            "content_sha256": content_sha256,
            "dek": base64.b64encode(dek).decode("ascii"),
            "signature": base64.b64encode(signature).decode("ascii"),
            "signer_public_key": base64.b64encode(signer_public_key).decode("ascii"),
        },
        files={"artifact": ("artifact.xdeploy", ciphertext, "application/octet-stream")},
    )
    if response.status_code in (401, 403):
        raise AuthenticationError(f"publish denied: {_error_message(response)}")
    _raise_for_status(response, PublishError, "publish")
    data = response.json()
    return PublishResult(
        artifact_id=data["artifact_id"],
        project_id=data["project_id"],
        version=data["version"],
        content_sha256=data["content_sha256"],
        size_bytes=data["size_bytes"],
        created_at=data["created_at"],
    )