From 72f142e468fcceb2706c0f8018afdeddca19fc62 Mon Sep 17 00:00:00 2001
From: Peli de Halleux <4175913+pelikhan@users.noreply.github.com>
Date: Fri, 6 Nov 2026 19:34:38 -0800
Subject: [PATCH] Add injectable Azure services and local integration coverage
Inventory Bicep resource coverage, introduce owned Blob, ADX and managed-identity interfaces, and exercise production adapters with local simulators and protocol fixtures. Wire hermetic service cases into the Azure integration harness and CI.
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
---
.github/workflows/azure-local-integration.yml | 16 +-
package.json | 1 +
scripts/azure-local/README.md | 110 +++++++-
scripts/azure-local/azure-local.sh | 22 +-
server/internal/server/session_archive.go | 17 +-
server/internal/sessionarchiveblob/archive.go | 88 +++----
.../azurite_integration_test.go | 201 +++++++++++++++
.../sessionarchiveblob/conformance_test.go | 6 +-
server/internal/sessionarchiveblob/http.go | 74 +++++-
server/internal/sessionarchiveblob/native.go | 25 +-
.../native_conformance_test.go | 4 +-
.../sessionarchiveblob/native_test.go | 2 +-
.../sessionarchiveblob/protocol_test.go | 10 +-
server/internal/sessionarchiveblob/query.go | 38 ++-
.../internal/sessionarchiveblob/services.go | 54 ++++
.../sessionarchiveblob/services_test.go | 144 +++++++++++
server/internal/tokenbroker/vault.go | 40 ++-
.../tokenbroker/vault_integration_test.go | 172 +++++++++++++
specs/session-archive.md | 11 +
.../integration/azure-local-services.test.mjs | 238 ++++++++++++++++++
.../server-azure-service-coverage.test.mjs | 62 +++++
21 files changed, 1241 insertions(+), 94 deletions(-)
create mode 100644 server/internal/sessionarchiveblob/azurite_integration_test.go
create mode 100644 server/internal/sessionarchiveblob/services.go
create mode 100644 server/internal/sessionarchiveblob/services_test.go
create mode 100644 server/internal/tokenbroker/vault_integration_test.go
create mode 100644 tests/integration/azure-local-services.test.mjs
create mode 100644 tests/unit/server-azure-service-coverage.test.mjs
diff --git a/.github/workflows/azure-local-integration.yml b/.github/workflows/azure-local-integration.yml
index 2e31f83957..b55ff4d0c6 100644
--- a/.github/workflows/azure-local-integration.yml
+++ b/.github/workflows/azure-local-integration.yml
@@ -1,4 +1,4 @@
-name: Local Azure Functions integration
+name: Local Azure services integration
on:
push:
@@ -9,7 +9,15 @@ on:
- scripts/azure-local/**
- server/**
- tests/integration/azure-local-functions.test.mjs
+ - tests/integration/azure-local-services.test.mjs
+ - tests/unit/server-azure-*.test.mjs
+ - internal/cli/**
+ - internal/azcli/**
+ - go.mod
+ - go.sum
+ - package.json
- .github/workflows/cao.json
+ - .github/workflows/cao.azure.json
- "!dashboard/site/**/*.md"
- "!server/**/*.md"
workflow_dispatch:
@@ -20,7 +28,7 @@ permissions:
jobs:
azure-local:
- name: Functions, Azurite, Redis, and Postgres
+ name: Functions, Storage, Postgres and Azure service contracts
runs-on: ubuntu-latest
timeout-minutes: 20
steps:
@@ -34,8 +42,10 @@ jobs:
with:
go-version: 1.27.3
cache: false
+ - name: Prepare Go dependencies before hermetic service tests
+ run: go mod download && go -C server mod download
- name: Run local Azure integration
- run: ./scripts/azure-local/azure-local.sh run
+ run: npm run test:integration:azure-services
- name: Upload local Azure logs
if: ${{ always() && failure() }}
uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
diff --git a/package.json b/package.json
index 02e87e7a31..5a29e3f313 100644
--- a/package.json
+++ b/package.json
@@ -34,6 +34,7 @@
"test:integration:setup": "CAO_SETUP_INTEGRATION=1 go -C server test -race ./internal/simulator -run '^(TestCAO(Setup|HostedCLISetup)Integration|TestSetup)' -count=1 -timeout=10m -v",
"test:integration:setup-azure": "CAO_SETUP_AZURE_INTEGRATION=1 GOTOOLCHAIN=local GOPROXY=off GOSUMDB=off GOVCS='*:off' go -C server test -race ./internal/simulator -run '^(TestCAOSetupAzureIntegration|TestSetupAzure|TestLocalTransport)' -count=1 -timeout=10m -v",
"test:integration:azure-install-journey": "CAO_SETUP_AZURE_INTEGRATION=1 GOTOOLCHAIN=local GOPROXY=off GOSUMDB=off GOVCS='*:off' go -C server test -race ./internal/simulator -run '^TestCAOAzureInstallJourneyIntegration$' -count=1 -timeout=10m -v",
+ "test:integration:azure-services": "bash scripts/azure-local/azure-local.sh run",
"test:integration:actions-simulator-postgres": "CAO_ACTIONS_SIMULATOR_POSTGRES=1 go -C server test -race ./internal/admission ./internal/webhookgateway -run '^(TestLocalActionsPostgresIntegration|TestIssueServiceSimulatorGatewayPostgres|TestGitHubSimulator(Gateway|Discovery)Postgres)$' -count=1 -timeout=2m -v",
"test:stress:dashboard-backfill": "CAO_BACKFILL_STRESS=1 bash scripts/test-dashboard-backfill.sh",
"test:stress:monthly-budget": "CAO_MONTHLY_BUDGET_STRESS=1 go -C server test ./internal/postgresx -run '^TestMonthlyBudgetStress$' -count=1 -timeout=40m -v",
diff --git a/scripts/azure-local/README.md b/scripts/azure-local/README.md
index 24e39dbae5..7a8cdb3c6c 100644
--- a/scripts/azure-local/README.md
+++ b/scripts/azure-local/README.md
@@ -1,4 +1,4 @@
-# Local Azure Functions simulation
+# Local Azure service integration
This Linux-only harness runs the CAO Azure Functions hosting boundary against
Azurite and a real PostgreSQL container. It uses no Azure subscription, Azure login,
@@ -11,6 +11,12 @@ or managed PostgreSQL TLS; those remain real-Azure smoke-test concerns.
- Linux with Bash, Docker, Go, Node.js, Python 3, curl, `flock`, unzip, and
`setsid`
- A running Docker daemon
+- Cached Go dependencies for both the root Azure deployment CLI and Go server.
+ Populate these before entering the hermetic service tests:
+
+ ```bash
+ go mod download && go -C server mod download
+ ```
Azure Functions Core Tools 4.15.1 is downloaded on first use from its official
GitHub release and verified by SHA-256. PostgreSQL 16 and Azurite 3.37.0 images
@@ -19,9 +25,102 @@ are pinned by tag or digest.
Run the complete suite:
```bash
-./scripts/azure-local/azure-local.sh run
+npm run test:integration:azure-services
+```
+
+## Bicep service inventory and local equivalents
+
+The inventory includes all eight `server/azure/*.bicep` modules, existing
+resources, and optional collector/ADX resources. The inventory guard compares
+every resource type with the generated ARM template; a new service fails the
+guard until its local coverage and limitations are reviewed.
+
+| Bicep resource type | Local equivalent and separate case | Not established locally |
+| --- | --- | --- |
+| `Microsoft.Web/sites` | [Functions Core Tools](https://learn.microsoft.com/en-us/azure/azure-functions/functions-run-local); route forwarding, health/readiness, shared PostgreSQL rate limits and missing configuration | Azure edge forwarding, scale-in, platform timeouts |
+| `Microsoft.Web/serverfarms` | Bicep/ARM hosting-plan contract assertions; no Azure plan emulator | Premium capacity, billing, always-on guarantees |
+| `Microsoft.Storage/storageAccounts` | [Azurite](https://learn.microsoft.com/en-us/azure/storage/common/storage-use-azurite); separate signed Blob read/write/ETag/denial, Queue visibility/receipt/acknowledgement and Table entity/ETag cases | Entra/RBAC, managed storage durability; Table emulation is preview |
+| `Microsoft.DBforPostgreSQL/flexibleServers` | Real PostgreSQL 16; production canonical adapter publication/reopen/contract-drift case | Flexible Server management, managed TLS, backups, HA |
+| `Microsoft.DBforPostgreSQL/flexibleServers/databases` | Production PostgreSQL operational adapter conformance; sessions, CAS, admission, queue capacity and leases in isolated namespaces | Azure database provisioning and restore |
+| `Microsoft.ContainerRegistry/registries` | Existing Azure CLI protocol fixture and production deploy adapter digest-binding/read-back cases; no ACR emulator | Registry push/pull, managed builds, AcrPull authorization |
+| `Microsoft.App/managedEnvironments` | Bicep/ARM environment, subnet and logging contracts; no managed-environment emulator | Azure environment lifecycle and networking |
+| `Microsoft.App/containerApps` | Collector image, identity, secret and deployment contract cases; ordinary local collection adapters are not a Container Apps emulator | Managed scheduling, replica coordination and autoscaling |
+| `Microsoft.App/jobs` | Production deploy adapter job ordering, failure and cancellation cases through the Azure CLI protocol fixture; setup journey separately executes real provisioning/ingestion subprocesses | Azure scheduler, execution isolation and platform job cancellation |
+| `Microsoft.KeyVault/vaults` | Local HTTP identity/secret fixture through the production `KeyVault.HTTPClient` seam; secret reads, rotation, denial, unavailability and redirect rejection | Azure vault RBAC, purge protection and App Service reference resolution |
+| `Microsoft.KeyVault/vaults/secrets` | Same fixture; missing/empty secret failure, uncached rotation; separate deploy adapter unresolved-reference case | Actual Key Vault references and secret propagation |
+| `Microsoft.ManagedIdentity/userAssignedIdentities` | Separate App Service and IMDS HTTP cases, audience/client selection, expired tokens and identity rejection | Entra token issuance, identity assignment and Azure authorization |
+| `Microsoft.Authorization/roleAssignments` | Existing least-privilege Bicep/ARM role/scope assertions; no Azure RBAC emulator | Effective deployed permissions |
+| `Microsoft.OperationalInsights/workspaces` | Workspace/shared-role logging deployment contracts; no Log Analytics emulator | Workspace retention, query execution and receipt of Azure platform logs |
+| `Microsoft.Insights/components` | Template-driven local TLS OTLP receiver case decodes production trace/metric/log protobuf and checks failure isolation; **not** an Application Insights emulator | Azure ingestion, Functions host telemetry and managed alerts |
+| `Microsoft.Insights/diagnosticSettings` | Diagnostic destination/category deployment contracts; no Azure diagnostic-settings emulator | Delivery of actual platform diagnostics |
+| `Microsoft.Network/virtualNetworks` | Bicep/ARM subnet/delegation assertions; no Azure VNet emulator | Routing, NSG enforcement and isolation |
+| `Microsoft.Network/privateDnsZones` | Private DNS naming and resource-link contracts | Azure private DNS resolution |
+| `Microsoft.Network/privateDnsZones/virtualNetworkLinks` | Private DNS/VNet linkage contracts | Effective DNS registration and reachability |
+| `Microsoft.Network/privateEndpoints` | ADX opt-in private endpoint/subnet/group contracts | Private Link connectivity and approval |
+| `Microsoft.Network/privateEndpoints/privateDnsZoneGroups` | ADX private DNS zone association contracts | Actual endpoint DNS resolution |
+| `Microsoft.Kusto/clusters` | Optional [official Kusto emulator](https://learn.microsoft.com/en-us/azure/data-explorer/kusto-emulator-overview); existing real-engine functional integration suite | Entra, streaming/managed ingestion, Azure cluster management and retention enforcement |
+| `Microsoft.Kusto/clusters/databases` | Same optional real engine, isolated databases and generated KQL/mapping/query cases | Managed database durability, retention and Azure permissions |
+
+There is no supported Azure-wide local emulator. Protocol fixtures and
+deployment-contract assertions above are deliberately not reported as managed
+service integration. No production endpoint, HTTPS requirement, credentials or
+rollout authority is weakened for tests. Every directly accessed Azure resource crosses an injectable interface. Existing
+contracts are retained, and concrete Blob/ADX archive dependencies and the
+Key Vault managed-identity dependency are replaced with owned service contracts:
+
+| Direct access | Consumer interface | Production adapter / local integration |
+| --- | --- | --- |
+| ARM resources, ACR, Functions settings/restart, Container Apps jobs, role/provider/name checks | `azcli.Executor` in deployment dependencies | `azcli.Process`; existing strict Azure CLI protocol simulator and deploy tests |
+| PostgreSQL canonical data | `dashboardstore.Store` / `dashboardstore.Reader` | `postgresx.Store`; real PostgreSQL publication/reopen tests |
+| PostgreSQL operational state | `operational.Store` and its domain service interfaces | `operational/postgres`; real multi-client conformance |
+| Key Vault secret read | `tokenbroker.PrivateKeyReader` | `tokenbroker.KeyVault`; local HTTP fixture |
+| App Service/IMDS managed identity | `tokenbroker.ManagedIdentityTokenSource` | `tokenbroker.ManagedIdentity`; separate local HTTP identity cases |
+| ADX streaming export and trace verification | `adxingest.Sink` / `adxtrace.Sink` | `adxingest.Client`; existing protocol and optional real Kusto engine suites |
+| Optional private archive Blob objects, leases and pointer CAS | `sessionarchiveblob.BlobStore` | Owned Blob HTTP adapter; new real Azurite integration through the interface, including verified retries after `BlobAlreadyExists` |
+| Optional private archive ADX metadata/query access | `sessionarchiveblob.QueryService` | Owned bounded ADX HTTP adapter; injected HTTP fixture and separate Azure conformance |
+| Archive publication and hosted read consumer | `sessionarchive.Publisher` / hosted reader/publisher interfaces | Legacy and native archive adapters; injected-service publication/restart/withdrawal cases |
+| Go telemetry export | OpenTelemetry provider/exporter interfaces | Existing OTLP exporters and local TLS protobuf receiver |
+
+`NewWithServices` and `NewNativeWithServices` accept explicit archive services.
+Both retain the existing configuration, scope and acceptance gates. Missing
+services fail; unavailable adapters do not trigger another backend. Immutable
+verification, lease ownership, conditional pointer publication, bounded complete
+query results, cancellation and withdrawal/recovery semantics remain part of the
+owned contracts. Service interfaces contain domain-owned results and errors,
+not database handles, Azure SDK types or raw HTTP responses.
+
+Functions owns its platform Blob/Queue/Table accesses; CAO has no production
+Queue/Table client to wrap. Bicep-only plan, network, diagnostics and role
+resources are accessed through the deployment executor, not invented runtime
+clients. No universal Azure service object or additional deployment resources
+are introduced.
+
+The default service suite does **not** run or claim ADX engine acceptance.
+After reviewing Microsoft's license, run it separately:
+
+```bash
+ACCEPT_EULA=Y bash scripts/test-kusto-emulator.sh
```
+The license prohibits benchmarks. The official emulator requires Linux/x86
+with the documented CPU features. An unavailable or unselected engine is
+incomplete ADX evidence, not a passing alternative.
+
+For an already-running isolated local stack, service cases can run separately
+from Functions forwarding:
+
+```bash
+node tests/integration/azure-local-services.test.mjs PATH/TO/runtime.json
+node tests/integration/azure-local-functions.test.mjs PATH/TO/runtime.json
+```
+
+The service runner requires explicit numeric-loopback endpoints, fresh
+synthetic resources and namespaces, and cached Go dependencies. It disables
+Go downloads, selects tests explicitly, rejects skipped or empty Go test
+selections, and fails on missing/unavailable services rather than falling back.
+Storage REST requests are test clients for the Functions platform dependency;
+they are not an additional production storage adapter.
+
The dispatcher also supports iterative use:
```bash
@@ -75,11 +174,12 @@ Before changing this harness:
- confirm all local settings are synthetic and no workflow secret or Azure
credential is required.
-## First end-to-end contract
+## Functions end-to-end contract
The integration test enters through the real Functions HTTP surface. It proves
that Core Tools registers and forwards the CAO routes, health/readiness reach
shared server logic and PostgreSQL, and PostgreSQL-backed pre-authentication
rate-limit state survives separate HTTP requests. Missing reviewed host policy
-still makes the custom handler fail predictably. Azurite is exercised only as
-Functions runtime storage.
+still makes the custom handler fail predictably. Separate service cases exercise
+Azurite Blob, Queue and Table before entering the Functions surface. Each case
+cleans up its own synthetic container, queue, table, database schema or namespace.
diff --git a/scripts/azure-local/azure-local.sh b/scripts/azure-local/azure-local.sh
index fd557ed1d8..b6aab43d6f 100755
--- a/scripts/azure-local/azure-local.sh
+++ b/scripts/azure-local/azure-local.sh
@@ -7,7 +7,7 @@ LOG_DIR="$STATE_DIR/logs"
APP_DIR="$STATE_DIR/function-app"
RUNTIME_ENV="$STATE_DIR/runtime.env"
RUNTIME_JSON="$STATE_DIR/runtime.json"
-HARNESS_VERSION=4
+HARNESS_VERSION=5
CORE_TOOLS_VERSION=4.15.1
CORE_TOOLS_SHA256=9b982efb4047c717c9b1c26beb2269c2ac41b29bf8d21f4df1130c1586cc81cd
POSTGRES_IMAGE=postgres:16-alpine
@@ -396,7 +396,15 @@ JSON
{
"stateDirectory": "$STATE_DIR",
"applicationDirectory": "$APP_DIR",
- "baseUrl": "http://127.0.0.1:$function_port"
+ "baseUrl": "http://127.0.0.1:$function_port",
+ "postgresUrl": "$CAO_POSTGRES_URL",
+ "storage": {
+ "account": "$AZURITE_ACCOUNT",
+ "key": "$azurite_key",
+ "blobEndpoint": "http://127.0.0.1:$blob_port/$AZURITE_ACCOUNT",
+ "queueEndpoint": "http://127.0.0.1:$queue_port/$AZURITE_ACCOUNT",
+ "tableEndpoint": "http://127.0.0.1:$table_port/$AZURITE_ACCOUNT"
+ }
}
JSON
@@ -430,12 +438,10 @@ wait_for_stack() {
test_stack() {
load_runtime
- set +e
- node "$ROOT/tests/integration/azure-local-functions.test.mjs" "$RUNTIME_JSON" \
- > >(tee "$LOG_DIR/integration-test.log") 2>&1
- local status=$?
- set -e
- return "$status"
+ run_logged "Azure service integration" service-integration-test.log \
+ node "$ROOT/tests/integration/azure-local-services.test.mjs" "$RUNTIME_JSON"
+ run_logged "Azure Functions integration" integration-test.log \
+ node "$ROOT/tests/integration/azure-local-functions.test.mjs" "$RUNTIME_JSON"
}
logs() {
diff --git a/server/internal/server/session_archive.go b/server/internal/server/session_archive.go
index c1d908ded3..512ef7a6ec 100644
--- a/server/internal/server/session_archive.go
+++ b/server/internal/server/session_archive.go
@@ -86,10 +86,23 @@ type localSessionArchive struct {
type hostedSessionArchive struct {
*sessionarchive.Archive
- publisher *sessionarchiveblob.Archive
+ publisher hostedArchivePublisher
namespace string
}
+type hostedArchiveReader interface {
+ ReadTimeline(context.Context, sessionarchive.Scope, sessionarchive.TimelineRequest) (uint64, []sessionarchive.Event, *sessionarchiveblob.Reader, error)
+ ReadSummaries(context.Context, sessionarchive.Scope, string, int) (uint64, []sessionarchive.Summary, *sessionarchiveblob.Reader, error)
+}
+
+type hostedArchivePublisher interface {
+ sessionarchive.Publisher
+ hostedArchiveReader
+ WithdrawRepository(context.Context, string) error
+ Maintain(context.Context) error
+ Close() error
+}
+
func (archive *hostedSessionArchive) Query(ctx context.Context, input sessionArchiveQuery) (sessionArchiveResult, error) {
return hostedArchiveQuery(ctx, archive.publisher, archive.namespace, input)
}
@@ -183,7 +196,7 @@ func (*nativeSessionArchive) Capture(context.Context, string, string, string) er
return collectnative.ErrUnsupported
}
-func hostedArchiveQuery(ctx context.Context, publisher *sessionarchiveblob.Archive, namespace string, input sessionArchiveQuery) (sessionArchiveResult, error) {
+func hostedArchiveQuery(ctx context.Context, publisher hostedArchiveReader, namespace string, input sessionArchiveQuery) (sessionArchiveResult, error) {
if err := input.validate(); err != nil {
return sessionArchiveResult{}, err
}
diff --git a/server/internal/sessionarchiveblob/archive.go b/server/internal/sessionarchiveblob/archive.go
index 031d4b9778..b8a67af96d 100644
--- a/server/internal/sessionarchiveblob/archive.go
+++ b/server/internal/sessionarchiveblob/archive.go
@@ -7,7 +7,6 @@ import (
"errors"
"fmt"
"io"
- "net/http"
"os"
"reflect"
"slices"
@@ -62,8 +61,8 @@ type Archive struct {
native bool
config Config
converter fileVerifier
- blob *blobClient
- queryHTTP *http.Client
+ blob BlobStore
+ queries QueryService
verifyADX bool
mu sync.Mutex
state archiveState
@@ -81,6 +80,16 @@ type Archive struct {
// New acquires the Blob lease only after the explicit activation gate. A
// disabled configuration returns an inert adapter without credential or I/O.
func New(ctx context.Context, config Config, converter *sessionarchive.Converter) (*Archive, error) {
+ return newArchive(ctx, config, converter, nil)
+}
+
+// NewWithServices preserves activation and destination gates while selecting
+// explicit adapters. It never substitutes services after an adapter failure.
+func NewWithServices(ctx context.Context, config Config, converter *sessionarchive.Converter, services Services) (*Archive, error) {
+ return newArchive(ctx, config, converter, &services)
+}
+
+func newArchive(ctx context.Context, config Config, converter *sessionarchive.Converter, services *Services) (*Archive, error) {
if err := config.Validate(); err != nil {
return nil, err
}
@@ -93,10 +102,15 @@ func New(ctx context.Context, config Config, converter *sessionarchive.Converter
if converter == nil {
return nil, errors.New("hosted archive requires the shared verifying Parquet converter")
}
- a := &Archive{config: config, converter: converter,
- blob: &blobClient{root: config.blobRoot(), http: newHTTP(config, "https://storage.azure.com/.default")},
- queryHTTP: newHTTP(config, "https://kusto.kusto.windows.net/.default"),
- verifyADX: true, now: time.Now}
+ if services == nil {
+ resolved := azureServices(config)
+ services = &resolved
+ }
+ if err := services.validate(); err != nil {
+ return nil, err
+ }
+ a := &Archive{config: config, converter: converter, blob: services.Blob,
+ queries: services.Queries, verifyADX: true, now: time.Now}
if err := a.open(ctx); err != nil {
return nil, err
}
@@ -104,33 +118,29 @@ func New(ctx context.Context, config Config, converter *sessionarchive.Converter
}
func (a *Archive) open(ctx context.Context) error {
- if err := a.blob.private(ctx); err != nil {
+ if err := a.blob.CheckPrivate(ctx); err != nil {
return err
}
initial, _ := json.Marshal(pointer{Version: 1})
- response, err := a.blob.request(ctx, http.MethodPut, "active.json", "", bytes.NewReader(initial), int64(len(initial)),
- http.Header{"If-None-Match": {"*"}, "x-ms-blob-type": {"BlockBlob"}})
- if response != nil {
- _ = response.Body.Close()
- }
- if err != nil && !isStatus(err, http.StatusPreconditionFailed) {
+ err := a.blob.CreatePointer(ctx, initial)
+ if err != nil && !errors.Is(err, ErrObjectExists) {
// An ambiguous create must be resolved by the same subsequent read.
- if _, _, readErr := a.blob.get(ctx, "active.json", 4096); readErr != nil {
+ if _, _, readErr := a.blob.Read(ctx, "active.json", 4096); readErr != nil {
return err
}
}
a.lease = uuid.NewString()
- if err := a.blob.lease(ctx, "acquire", a.lease); err != nil {
+ if err := a.blob.Lease(ctx, "acquire", a.lease); err != nil {
return err
}
if err := a.load(ctx); err != nil {
- _ = a.blob.lease(ctx, "release", a.lease)
+ _ = a.blob.Lease(ctx, "release", a.lease)
return err
}
// No runtime DDL: prove the preprovisioned external table is the approved
// immutable-object partitioned destination before accepting any data.
if err := a.verifyTable(ctx); err != nil {
- _ = a.blob.lease(ctx, "release", a.lease)
+ _ = a.blob.Lease(ctx, "release", a.lease)
return err
}
lifetime, cancel := context.WithCancel(context.WithoutCancel(ctx))
@@ -152,7 +162,7 @@ func decode(data []byte, target any) error {
}
func (a *Archive) load(ctx context.Context) error {
- data, etag, err := a.blob.get(ctx, "active.json", 4096)
+ data, etag, err := a.blob.Read(ctx, "active.json", 4096)
if err != nil {
return err
}
@@ -166,7 +176,7 @@ func (a *Archive) load(ctx context.Context) error {
if !digestPattern.MatchString(active.State) {
return ErrIntegrity
}
- data, _, err = a.blob.get(ctx, "manifests/"+active.State+".json", a.config.MaxStateBytes)
+ data, _, err = a.blob.Read(ctx, "manifests/"+active.State+".json", a.config.MaxStateBytes)
if err != nil || digest(data) != active.State || decode(data, &state) != nil ||
state.Manifest.Generation != active.Generation {
return ErrIntegrity
@@ -192,7 +202,7 @@ func (a *Archive) renew(ctx context.Context) {
return
case <-ticker.C:
step, cancel := context.WithTimeout(ctx, 10*time.Second)
- err := a.blob.lease(step, "renew", a.lease)
+ err := a.blob.Lease(step, "renew", a.lease)
cancel()
if err != nil {
// Renewal must not wait behind a long conversion or upload.
@@ -239,7 +249,7 @@ func (a *Archive) save(ctx context.Context, state archiveState, horizon time.Tim
if err := a.validateState(state); err != nil {
return err
}
- if err := a.blob.lease(ctx, "renew", a.lease); err != nil {
+ if err := a.blob.Lease(ctx, "renew", a.lease); err != nil {
a.failLease()
return err
}
@@ -248,21 +258,17 @@ func (a *Archive) save(ctx context.Context, state archiveState, horizon time.Tim
return ErrCapacity
}
hash := digest(data)
- if err := a.blob.immutable(ctx, "manifests/"+hash+".json", bytes.NewReader(data), int64(len(data)), hash); err != nil {
+ if err := a.blob.WriteImmutable(ctx, "manifests/"+hash+".json", bytes.NewReader(data), int64(len(data)), hash); err != nil {
return err
}
next := pointer{Version: 1, State: hash, Generation: state.Manifest.Generation, Revisions: a.active.Revisions + 1,
ReadersUntil: maxTime(a.active.ReadersUntil, horizon)}
content, _ := json.Marshal(next)
- response, err := a.blob.request(ctx, http.MethodPut, "active.json", "", bytes.NewReader(content), int64(len(content)),
- http.Header{"If-Match": {a.etag}, "x-ms-lease-id": {a.lease}, "x-ms-blob-type": {"BlockBlob"}})
- if response != nil {
- _ = response.Body.Close()
- }
+ etag, err := a.blob.ComparePointer(ctx, content, a.etag, a.lease)
if err != nil {
// CAS may have committed despite a lost acknowledgement. Resolve it
// under our still-held lease; never assume success on a failed read.
- check, etag, readErr := a.blob.get(ctx, "active.json", 4096)
+ check, etag, readErr := a.blob.Read(ctx, "active.json", 4096)
var current pointer
if readErr == nil && decode(check, ¤t) == nil &&
current.Version == next.Version && current.State == next.State && current.Generation == next.Generation &&
@@ -270,13 +276,12 @@ func (a *Archive) save(ctx context.Context, state archiveState, horizon time.Tim
a.active, a.etag, a.state = next, etag, state
return nil
}
- if isStatus(err, http.StatusPreconditionFailed) || isStatus(err, http.StatusConflict) {
+ if errors.Is(err, ErrConflict) {
a.failLease()
return ErrConflict
}
return err
}
- etag := response.Header.Get("ETag")
if etag == "" {
a.failLease()
return ErrIntegrity
@@ -468,7 +473,7 @@ func (a *Archive) uploadFile(ctx context.Context, file sessionarchive.File) erro
if err != nil || !info.Mode().IsRegular() || info.Size() != file.Bytes {
return ErrIntegrity
}
- return a.blob.immutable(ctx, objectKey(file), handle, file.Bytes, file.Digest)
+ return a.blob.WriteImmutable(ctx, objectKey(file), handle, file.Bytes, file.Digest)
}
// The caller holds the lease owner's mutex. Both producer profiles share the
@@ -487,7 +492,7 @@ func (a *Archive) publish(ctx context.Context, candidate sessionarchive.Manifest
if normalized.Generation != a.state.Manifest.Generation+1 {
return ErrConflict
}
- if err := a.blob.private(ctx); err != nil {
+ if err := a.blob.CheckPrivate(ctx); err != nil {
return err
}
if a.verifyADX {
@@ -610,12 +615,12 @@ func (a *Archive) publish(ctx context.Context, candidate sessionarchive.Manifest
if !reflect.DeepEqual(previous, proposed) {
return ErrIntegrity
}
- if err := a.blob.verify(ctx, previous.Path, previous.Digest, previous.Bytes); err != nil {
+ if err := a.blob.Verify(ctx, previous.Path, previous.Digest, previous.Bytes); err != nil {
return err
}
continue
}
- if err := a.blob.lease(ctx, "renew", a.lease); err != nil {
+ if err := a.blob.Lease(ctx, "renew", a.lease); err != nil {
a.failLease()
return err
}
@@ -658,7 +663,7 @@ func (a *Archive) ResolvePending(ctx context.Context, candidate sessionarchive.M
if err != nil {
return false, err
}
- if err := a.blob.lease(ctx, "renew", a.lease); err != nil {
+ if err := a.blob.Lease(ctx, "renew", a.lease); err != nil {
a.failLease()
return false, err
}
@@ -690,15 +695,12 @@ func (a *Archive) Maintain(ctx context.Context) error {
if a.now().Before(r.After) || a.now().Before(a.active.ReadersUntil.Add(a.config.RetirementGrace)) {
continue
}
- if err := a.blob.lease(ctx, "renew", a.lease); err != nil {
+ if err := a.blob.Lease(ctx, "renew", a.lease); err != nil {
a.failLease()
return err
}
- response, err := a.blob.request(ctx, http.MethodDelete, r.Key, "", nil, 0, http.Header{})
- if response != nil {
- _ = response.Body.Close()
- }
- if err != nil && !isStatus(err, http.StatusNotFound) {
+ err := a.blob.Delete(ctx, r.Key)
+ if err != nil && !errors.Is(err, ErrNotFound) {
return err
}
next := a.state
@@ -802,7 +804,7 @@ func (a *Archive) Close() error {
<-a.done
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
- return a.blob.lease(ctx, "release", a.lease)
+ return a.blob.Lease(ctx, "release", a.lease)
}
func (a *Archive) String() string {
diff --git a/server/internal/sessionarchiveblob/azurite_integration_test.go b/server/internal/sessionarchiveblob/azurite_integration_test.go
new file mode 100644
index 0000000000..b8a1ebb196
--- /dev/null
+++ b/server/internal/sessionarchiveblob/azurite_integration_test.go
@@ -0,0 +1,201 @@
+package sessionarchiveblob
+
+import (
+ "bytes"
+ "context"
+ "crypto/hmac"
+ "crypto/rand"
+ "crypto/sha256"
+ "encoding/base64"
+ "errors"
+ "net/http"
+ "net/url"
+ "os"
+ "slices"
+ "strconv"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/google/uuid"
+
+ "github.com/githubnext/gh-cao/server/internal/simulator"
+)
+
+type azuriteTransport struct {
+ local *simulator.LocalTransport
+ target *url.URL
+ account string
+ key []byte
+}
+
+func (a *azuriteTransport) RoundTrip(request *http.Request) (*http.Response, error) {
+ if request.URL.Scheme != "https" || request.URL.Host != a.account+".blob.core.windows.net" || request.URL.User != nil {
+ return nil, errors.New("Azurite fixture blocked an unapproved Blob authority")
+ }
+ mapped := request.Clone(request.Context())
+ target := *request.URL
+ target.Scheme, target.Host = a.target.Scheme, a.target.Host
+ target.Path = "/" + a.account + request.URL.Path
+ mapped.URL, mapped.Host = &target, target.Host
+ mapped.Header.Set("x-ms-date", time.Now().UTC().Format(http.TimeFormat))
+ mapped.Header.Set("x-ms-version", "2023-11-03")
+ length := ""
+ if mapped.ContentLength > 0 {
+ length = strconv.FormatInt(mapped.ContentLength, 10)
+ }
+ standard := strings.Join([]string{mapped.Method,
+ mapped.Header.Get("Content-Encoding"), mapped.Header.Get("Content-Language"), length,
+ mapped.Header.Get("Content-MD5"), mapped.Header.Get("Content-Type"), "",
+ mapped.Header.Get("If-Modified-Since"), mapped.Header.Get("If-Match"),
+ mapped.Header.Get("If-None-Match"), mapped.Header.Get("If-Unmodified-Since"), mapped.Header.Get("Range"),
+ }, "\n")
+ var names []string
+ azureHeaders := map[string]string{}
+ for name, values := range mapped.Header {
+ if strings.HasPrefix(strings.ToLower(name), "x-ms-") {
+ name = strings.ToLower(name)
+ names = append(names, name)
+ azureHeaders[name] = strings.Join(values, ",")
+ }
+ }
+ slices.Sort(names)
+ var canonical strings.Builder
+ canonical.WriteString(standard + "\n")
+ for _, name := range names {
+ canonical.WriteString(name + ":" + azureHeaders[name] + "\n")
+ }
+ canonical.WriteString("/" + a.account + mapped.URL.Path)
+ names = names[:0]
+ for name := range mapped.URL.Query() {
+ names = append(names, name)
+ }
+ slices.Sort(names)
+ for _, name := range names {
+ values := mapped.URL.Query()[name]
+ slices.Sort(values)
+ canonical.WriteString("\n" + name + ":" + strings.Join(values, ","))
+ }
+ signature := hmac.New(sha256.New, a.key)
+ _, _ = signature.Write([]byte(canonical.String()))
+ mapped.Header.Set("Authorization", "SharedKey "+a.account+":"+base64.StdEncoding.EncodeToString(signature.Sum(nil)))
+ return a.local.RoundTrip(mapped)
+}
+
+func TestAzuriteBlobStoreIntegration(t *testing.T) {
+ endpoint := os.Getenv("CAO_AZURITE_BLOB_URL")
+ if endpoint == "" {
+ t.Skip("requires official Azurite; run npm run test:integration:azure-services")
+ }
+ account := os.Getenv("CAO_AZURITE_ACCOUNT")
+ if !accountPattern.MatchString(account) {
+ t.Fatal("Azurite requires an explicit synthetic account")
+ }
+ target, err := url.Parse(endpoint)
+ if err != nil || target.Path != "/"+account {
+ t.Fatal("Azurite requires the exact path-style synthetic account endpoint")
+ }
+ local, err := simulator.NewLocalTransport(endpoint)
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(local.CloseIdleConnections)
+ key, err := base64.StdEncoding.DecodeString(os.Getenv("CAO_AZURITE_KEY"))
+ if err != nil || len(key) != 32 {
+ t.Fatal("Azurite requires a synthetic 32-byte account key")
+ }
+ client := &http.Client{Transport: &azuriteTransport{local: local, target: target, account: account, key: key},
+ Timeout: 5 * time.Second, CheckRedirect: rejectRedirect}
+ container := "cao-adapter-" + strings.ToLower(rand.Text())
+ root := "https://" + account + ".blob.core.windows.net/" + container
+ containerRequest := func(method string) {
+ t.Helper()
+ ctx, cancel := context.WithTimeout(context.WithoutCancel(t.Context()), 5*time.Second)
+ defer cancel()
+ request, err := http.NewRequestWithContext(ctx, method, root+"?restype=container", nil) // #nosec G704 -- validated synthetic Blob authority is mapped only to the exact numeric-loopback emulator.
+ if err != nil {
+ t.Fatal(err)
+ }
+ response, err := client.Do(request) // #nosec G704 -- azuriteTransport rejects other authorities and LocalTransport blocks non-loopback connections.
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer func() { _ = response.Body.Close() }()
+ if response.StatusCode < 200 || response.StatusCode >= 300 {
+ t.Fatalf("Azurite container operation returned HTTP %d", response.StatusCode)
+ }
+ }
+ containerRequest(http.MethodPut)
+ t.Cleanup(func() { containerRequest(http.MethodDelete) })
+ var store BlobStore = &blobClient{root: root + "/archive", http: client}
+ t.Run("private immutable verified objects", func(t *testing.T) {
+ if err := store.CheckPrivate(t.Context()); err != nil {
+ t.Fatal(err)
+ }
+ data := []byte("synthetic immutable evidence")
+ for range 2 {
+ if err := store.WriteImmutable(t.Context(), "objects/evidence", bytes.NewReader(data), int64(len(data)), digest(data)); err != nil {
+ t.Fatal(err)
+ }
+ }
+ read, etag, err := store.Read(t.Context(), "objects/evidence", 4096)
+ if err != nil || !bytes.Equal(read, data) || etag == "" {
+ t.Fatalf("read verified object failed: %v", err)
+ }
+ if err := store.Verify(t.Context(), "objects/evidence", digest([]byte("wrong")), int64(len(data))); !errors.Is(err, ErrIntegrity) {
+ t.Fatalf("corrupt digest accepted: %v", err)
+ }
+ if _, _, err := store.Read(t.Context(), "missing", 4096); !errors.Is(err, ErrNotFound) {
+ t.Fatalf("missing object did not return domain error: %v", err)
+ }
+ })
+ t.Run("leased pointer CAS", func(t *testing.T) {
+ if err := store.CreatePointer(t.Context(), []byte("{}")); err != nil {
+ t.Fatal(err)
+ }
+ if err := store.CreatePointer(t.Context(), []byte("{}")); !errors.Is(err, ErrObjectExists) {
+ t.Fatalf("pointer create was not exclusive: %v", err)
+ }
+ lease := uuid.NewString()
+ if err := store.Lease(t.Context(), "acquire", lease); err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() {
+ if err := store.Lease(context.WithoutCancel(t.Context()), "release", lease); err != nil {
+ t.Error(err)
+ }
+ })
+ _, original, err := store.Read(t.Context(), "active.json", 4096)
+ if err != nil {
+ t.Fatal(err)
+ }
+ data := []byte(`{"generation":1}`)
+ etag, err := store.ComparePointer(t.Context(), data, original, lease)
+ if err != nil || etag == "" || etag == original {
+ t.Fatalf("pointer CAS failed: %v", err)
+ }
+ if _, err := store.ComparePointer(t.Context(), []byte("{}"), original, lease); !errors.Is(err, ErrConflict) {
+ t.Fatalf("stale pointer CAS did not conflict: %v", err)
+ }
+ read, _, err := store.Read(t.Context(), "active.json", 4096)
+ if err != nil || !bytes.Equal(read, data) {
+ t.Fatal("stale writer changed the committed pointer")
+ }
+ if err := store.Lease(t.Context(), "renew", lease); err != nil {
+ t.Fatal(err)
+ }
+ })
+ t.Run("delete and cancellation", func(t *testing.T) {
+ if err := store.Delete(t.Context(), "objects/evidence"); err != nil {
+ t.Fatal(err)
+ }
+ if err := store.Delete(t.Context(), "objects/evidence"); !errors.Is(err, ErrNotFound) {
+ t.Fatalf("repeated delete did not report missing object: %v", err)
+ }
+ ctx, cancel := context.WithCancel(t.Context())
+ cancel()
+ if _, _, err := store.Read(ctx, "active.json", 4096); err == nil {
+ t.Fatal("cancelled Blob read succeeded")
+ }
+ })
+}
diff --git a/server/internal/sessionarchiveblob/conformance_test.go b/server/internal/sessionarchiveblob/conformance_test.go
index 8453ae422b..719d5c897b 100644
--- a/server/internal/sessionarchiveblob/conformance_test.go
+++ b/server/internal/sessionarchiveblob/conformance_test.go
@@ -481,7 +481,7 @@ func TestAzureArchiveConformance(t *testing.T) {
if err := a.Maintain(t.Context()); err != nil {
t.Fatal(err)
}
- if err := a.blob.verify(t.Context(), objectKey(corrected), corrected.Digest, corrected.Bytes); err != nil {
+ if err := a.blob.Verify(t.Context(), objectKey(corrected), corrected.Digest, corrected.Bytes); err != nil {
t.Fatal("live retirement deleted an object before the persisted recovery grace")
}
deadline := a.active.ReadersUntil.Add(config.RetirementGrace + time.Second)
@@ -495,10 +495,10 @@ func TestAzureArchiveConformance(t *testing.T) {
if err := a.Maintain(t.Context()); err != nil {
t.Fatal(err)
}
- if err := a.blob.verify(t.Context(), objectKey(corrected), corrected.Digest, corrected.Bytes); !isStatus(err, http.StatusNotFound) {
+ if err := a.blob.Verify(t.Context(), objectKey(corrected), corrected.Digest, corrected.Bytes); !isStatus(err, http.StatusNotFound) {
t.Fatal("live eligible retired object was not physically deleted")
}
- if err := a.blob.verify(t.Context(), objectKey(original), original.Digest, original.Bytes); err != nil {
+ if err := a.blob.Verify(t.Context(), objectKey(original), original.Digest, original.Bytes); err != nil {
t.Fatal("live retirement deleted a currently selected packed file")
}
t.Log("Azure destination conformance passed; explicit runtime acceptance binding:", config.Acceptance)
diff --git a/server/internal/sessionarchiveblob/http.go b/server/internal/sessionarchiveblob/http.go
index afd7a0c221..385b6c66a2 100644
--- a/server/internal/sessionarchiveblob/http.go
+++ b/server/internal/sessionarchiveblob/http.go
@@ -1,6 +1,7 @@
package sessionarchiveblob
import (
+ "bytes"
"context"
"crypto/sha256"
"encoding/hex"
@@ -67,6 +68,30 @@ type statusError int
func (s statusError) Error() string { return fmt.Sprintf("archive Blob returned HTTP %d", s) }
+type blobResponseError struct {
+ status statusError
+ code string
+}
+
+func (e *blobResponseError) Error() string { return e.status.Error() }
+func (e *blobResponseError) Unwrap() error { return e.status }
+func (e *blobResponseError) Is(target error) bool {
+ return target == ErrObjectExists && int(e.status) == http.StatusConflict && e.code == "BlobAlreadyExists"
+}
+
+func (s statusError) Is(target error) bool {
+ switch target {
+ case ErrObjectExists:
+ return int(s) == http.StatusPreconditionFailed
+ case ErrNotFound:
+ return int(s) == http.StatusNotFound
+ case ErrConflict:
+ return int(s) == http.StatusPreconditionFailed || int(s) == http.StatusConflict
+ default:
+ return false
+ }
+}
+
func isStatus(err error, status int) bool {
var e statusError
return errors.As(err, &e) && int(e) == status
@@ -98,12 +123,12 @@ func (b *blobClient) request(ctx context.Context, method, key, query string, bod
}
if response.StatusCode < 200 || response.StatusCode >= 300 {
_ = response.Body.Close()
- return nil, statusError(response.StatusCode)
+ return nil, &blobResponseError{status: statusError(response.StatusCode), code: response.Header.Get("x-ms-error-code")}
}
return response, nil
}
-func (b *blobClient) private(ctx context.Context) error {
+func (b *blobClient) CheckPrivate(ctx context.Context) error {
u, err := url.Parse(b.root)
if err != nil {
return ErrScope
@@ -122,7 +147,7 @@ func (b *blobClient) private(ctx context.Context) error {
return nil
}
-func (b *blobClient) get(ctx context.Context, key string, max int64) ([]byte, string, error) {
+func (b *blobClient) Read(ctx context.Context, key string, max int64) ([]byte, string, error) {
response, err := b.request(ctx, http.MethodGet, key, "", nil, 0, http.Header{})
if err != nil {
return nil, "", err
@@ -144,7 +169,7 @@ func digest(data []byte) string {
return hex.EncodeToString(sum[:])
}
-func (b *blobClient) verify(ctx context.Context, key, expected string, size int64) error {
+func (b *blobClient) Verify(ctx context.Context, key, expected string, size int64) error {
response, err := b.request(ctx, http.MethodGet, key, "", nil, 0, http.Header{})
if err != nil {
return err
@@ -158,11 +183,11 @@ func (b *blobClient) verify(ctx context.Context, key, expected string, size int6
return nil
}
-func (b *blobClient) immutable(ctx context.Context, key string, body io.Reader, size int64, expected string) error {
+func (b *blobClient) WriteImmutable(ctx context.Context, key string, body io.Reader, size int64, expected string) error {
headers := http.Header{"If-None-Match": {"*"}, "x-ms-blob-type": {"BlockBlob"},
"Content-Type": {"application/octet-stream"}}
response, err := b.request(ctx, http.MethodPut, key, "", body, size, headers)
- if err != nil && !isStatus(err, http.StatusPreconditionFailed) {
+ if err != nil && !errors.Is(err, ErrObjectExists) {
return err
}
if response != nil {
@@ -170,10 +195,10 @@ func (b *blobClient) immutable(ctx context.Context, key string, body io.Reader,
}
// Includes deduplication after an ambiguous upload acknowledgement. Never
// trust object names or a caller-provided metadata digest alone.
- return b.verify(ctx, key, expected, size)
+ return b.Verify(ctx, key, expected, size)
}
-func (b *blobClient) lease(ctx context.Context, action, id string) error {
+func (b *blobClient) Lease(ctx context.Context, action, id string) error {
headers := http.Header{"x-ms-lease-action": {action}}
if action == "acquire" {
headers.Set("x-ms-lease-duration", "60")
@@ -192,3 +217,36 @@ func (b *blobClient) lease(ctx context.Context, action, id string) error {
}
return nil
}
+
+func (b *blobClient) CreatePointer(ctx context.Context, data []byte) error {
+ response, err := b.request(ctx, http.MethodPut, "active.json", "", bytes.NewReader(data), int64(len(data)),
+ http.Header{"If-None-Match": {"*"}, "x-ms-blob-type": {"BlockBlob"}})
+ if response != nil {
+ _ = response.Body.Close()
+ }
+ return err
+}
+
+func (b *blobClient) ComparePointer(ctx context.Context, data []byte, etag, lease string) (string, error) {
+ if etag == "" {
+ return "", ErrConflict
+ }
+ if lease == "" {
+ return "", ErrLease
+ }
+ response, err := b.request(ctx, http.MethodPut, "active.json", "", bytes.NewReader(data), int64(len(data)),
+ http.Header{"If-Match": {etag}, "x-ms-lease-id": {lease}, "x-ms-blob-type": {"BlockBlob"}})
+ if err != nil {
+ return "", err
+ }
+ defer func() { _ = response.Body.Close() }()
+ return response.Header.Get("ETag"), nil
+}
+
+func (b *blobClient) Delete(ctx context.Context, key string) error {
+ response, err := b.request(ctx, http.MethodDelete, key, "", nil, 0, http.Header{})
+ if response != nil {
+ _ = response.Body.Close()
+ }
+ return err
+}
diff --git a/server/internal/sessionarchiveblob/native.go b/server/internal/sessionarchiveblob/native.go
index 797df4cad3..1458ae4a4f 100644
--- a/server/internal/sessionarchiveblob/native.go
+++ b/server/internal/sessionarchiveblob/native.go
@@ -31,6 +31,14 @@ type Native struct {
}
func NewNative(ctx context.Context, capture sessionarchive.Config, destination Config) (*Native, error) {
+ return newNative(ctx, capture, destination, nil)
+}
+
+func NewNativeWithServices(ctx context.Context, capture sessionarchive.Config, destination Config, services Services) (*Native, error) {
+ return newNative(ctx, capture, destination, &services)
+}
+
+func newNative(ctx context.Context, capture sessionarchive.Config, destination Config, services *Services) (*Native, error) {
if !capture.Enabled || !capture.NativeProducer || !destination.Enabled {
return nil, errors.New("native archive requires explicit approved capture and Blob selection")
}
@@ -46,10 +54,15 @@ func NewNative(ctx context.Context, capture sessionarchive.Config, destination C
if destination.Acceptance != destination.NativeAcceptanceBinding() {
return nil, ErrConformance
}
- a := &Archive{native: true, config: destination,
- blob: &blobClient{root: destination.blobRoot(), http: newHTTP(destination, "https://storage.azure.com/.default")},
- queryHTTP: newHTTP(destination, "https://kusto.kusto.windows.net/.default"),
- verifyADX: true, now: time.Now}
+ if services == nil {
+ resolved := azureServices(destination)
+ services = &resolved
+ }
+ if err := services.validate(); err != nil {
+ return nil, err
+ }
+ a := &Archive{native: true, config: destination, blob: services.Blob,
+ queries: services.Queries, verifyADX: true, now: time.Now}
if err := a.open(ctx); err != nil {
return nil, err
}
@@ -219,7 +232,7 @@ func (n *Native) captureSession(ctx context.Context, source sessionarchive.Sourc
if index >= 0 && records[index].Gap == "" && records[index].Member == m {
for _, file := range n.state.Manifest.Files {
if slices.Contains(file.Members, m) {
- return n.blob.verify(ctx, file.Path, file.Digest, file.Bytes)
+ return n.blob.Verify(ctx, file.Path, file.Digest, file.Bytes)
}
}
return ErrIntegrity
@@ -268,7 +281,7 @@ func (n *Native) captureSession(ctx context.Context, source sessionarchive.Sourc
if f.Digest != file.Digest || f.Bytes != int64(len(data)) || digest(data) != f.Digest {
return ErrIntegrity
}
- return n.blob.immutable(ctx, objectKey(f), bytes.NewReader(data), f.Bytes, f.Digest)
+ return n.blob.WriteImmutable(ctx, objectKey(f), bytes.NewReader(data), f.Bytes, f.Digest)
}, records, nil)
}
diff --git a/server/internal/sessionarchiveblob/native_conformance_test.go b/server/internal/sessionarchiveblob/native_conformance_test.go
index b857833210..5d46aee0b8 100644
--- a/server/internal/sessionarchiveblob/native_conformance_test.go
+++ b/server/internal/sessionarchiveblob/native_conformance_test.go
@@ -112,7 +112,7 @@ func TestAzureNativeArchiveConformance(t *testing.T) {
if err := n.Maintain(t.Context()); err != nil {
t.Fatal(err)
}
- if err := n.blob.verify(t.Context(), originalFile.Path, originalFile.Digest, originalFile.Bytes); err != nil {
+ if err := n.blob.Verify(t.Context(), originalFile.Path, originalFile.Digest, originalFile.Bytes); err != nil {
t.Fatal("native retirement deleted historical data before conservative grace")
}
timer := time.NewTimer(time.Until(n.active.ReadersUntil.Add(destination.RetirementGrace + time.Second)))
@@ -125,7 +125,7 @@ func TestAzureNativeArchiveConformance(t *testing.T) {
if err := n.Maintain(t.Context()); err != nil {
t.Fatal(err)
}
- if err := n.blob.verify(t.Context(), originalFile.Path, originalFile.Digest, originalFile.Bytes); !isStatus(err, http.StatusNotFound) {
+ if err := n.blob.Verify(t.Context(), originalFile.Path, originalFile.Digest, originalFile.Bytes); !isStatus(err, http.StatusNotFound) {
t.Fatal("native eligible object was not physically retired")
}
t.Log("Native Azure destination conformance passed; explicit runtime acceptance binding:", destination.Acceptance)
diff --git a/server/internal/sessionarchiveblob/native_test.go b/server/internal/sessionarchiveblob/native_test.go
index 990cff1294..e52b0cc79c 100644
--- a/server/internal/sessionarchiveblob/native_test.go
+++ b/server/internal/sessionarchiveblob/native_test.go
@@ -44,7 +44,7 @@ func restartNative(t *testing.T, previous *Native) *Native {
if err := previous.Close(); err != nil {
t.Fatal(err)
}
- a := &Archive{native: true, config: previous.config, blob: previous.blob, queryHTTP: previous.queryHTTP, now: previous.now}
+ a := &Archive{native: true, config: previous.config, blob: previous.blob, queries: previous.queries, now: previous.now}
if err := a.open(t.Context()); err != nil {
t.Fatal(err)
}
diff --git a/server/internal/sessionarchiveblob/protocol_test.go b/server/internal/sessionarchiveblob/protocol_test.go
index 229f33214b..fbacb5b511 100644
--- a/server/internal/sessionarchiveblob/protocol_test.go
+++ b/server/internal/sessionarchiveblob/protocol_test.go
@@ -238,8 +238,8 @@ func protocolArchive(t *testing.T, s *blobProtocol, endpoint string) *Archive {
c.Endpoint = endpoint // Only test construction permits numeric-loopback HTTP.
a := &Archive{config: c, converter: protocolVerifier{}, blob: &blobClient{root: endpoint + "/container/archive",
http: &http.Client{CheckRedirect: rejectRedirect}},
- queryHTTP: &http.Client{CheckRedirect: rejectRedirect},
- now: func() time.Time { s.mu.Lock(); defer s.mu.Unlock(); return s.now }}
+ queries: &adxQueryClient{config: c, http: &http.Client{CheckRedirect: rejectRedirect}},
+ now: func() time.Time { s.mu.Lock(); defer s.mu.Unlock(); return s.now }}
if err := a.open(t.Context()); err != nil {
t.Fatal(err)
}
@@ -281,7 +281,7 @@ func TestBlobImmutablePublicationRestartAndOverlap(t *testing.T) {
if bytes.Contains(s.objects["manifests/"+a.active.State+".json"], []byte(m.Files[0].Path)) {
t.Fatal("private staging path leaked into remote manifest")
}
- competing := &Archive{config: a.config, blob: a.blob, queryHTTP: a.queryHTTP, now: a.now}
+ competing := &Archive{config: a.config, blob: a.blob, queries: a.queries, now: a.now}
if err := competing.open(t.Context()); !errors.Is(err, ErrLease) {
t.Fatalf("restart overlap was not rejected: %v", err)
}
@@ -464,7 +464,7 @@ func TestBlobDefaultOffAndDestinationValidation(t *testing.T) {
func TestBlobPublicContainerAndRedirectRejected(t *testing.T) {
a, s, _ := newProtocol(t)
s.public = true
- if err := a.blob.private(t.Context()); err == nil {
+ if err := a.blob.CheckPrivate(t.Context()); err == nil {
t.Fatal("public container accepted")
}
redirected := false
@@ -475,7 +475,7 @@ func TestBlobPublicContainerAndRedirectRejected(t *testing.T) {
}))
defer source.Close()
b := blobClient{root: source.URL, http: &http.Client{CheckRedirect: rejectRedirect}}
- if _, _, err := b.get(t.Context(), "object", 4096); err == nil || redirected {
+ if _, _, err := b.Read(t.Context(), "object", 4096); err == nil || redirected {
t.Fatal("credential-bearing redirect followed")
}
}
diff --git a/server/internal/sessionarchiveblob/query.go b/server/internal/sessionarchiveblob/query.go
index c3104248e1..76481aad56 100644
--- a/server/internal/sessionarchiveblob/query.go
+++ b/server/internal/sessionarchiveblob/query.go
@@ -18,16 +18,44 @@ import (
"github.com/githubnext/gh-cao/server/internal/sessionarchive"
)
-type queryTable struct {
+type QueryResult struct {
TableName string
- Columns []struct{ ColumnName string }
+ Columns []QueryColumn
Rows [][]json.RawMessage
}
+type QueryColumn struct{ ColumnName string }
+
+type queryTable = QueryResult
+
+type adxQueryClient struct {
+ config Config
+ http *http.Client
+}
+
+func (c *adxQueryClient) Query(ctx context.Context, kql string) (QueryResult, error) {
+ return c.execute(ctx, "/v1/rest/query", kql)
+}
+
+func (c *adxQueryClient) Management(ctx context.Context, kql string) (QueryResult, error) {
+ return c.execute(ctx, "/v1/rest/mgmt", kql)
+}
+
+func (a *Archive) query(ctx context.Context, path, kql string) (queryTable, error) {
+ switch path {
+ case "/v1/rest/query":
+ return a.queries.Query(ctx, kql)
+ case "/v1/rest/mgmt":
+ return a.queries.Management(ctx, kql)
+ default:
+ return queryTable{}, ErrQuery
+ }
+}
+
// query rejects partial results, unrecognized auxiliary tables and server
// diagnostics. Raw Azure bodies (which may contain credentials or evidence)
// are never returned to callers.
-func (a *Archive) query(ctx context.Context, path, kql string) (queryTable, error) {
+func (a *adxQueryClient) execute(ctx context.Context, path, kql string) (queryTable, error) {
body, err := json.Marshal(map[string]any{"db": a.config.Database, "csl": kql,
"properties": map[string]any{"Options": map[string]any{
"servertimeout": fmt.Sprintf("00:%02d:%02d", int(a.config.QueryTimeout/time.Minute), int(a.config.QueryTimeout/time.Second)%60),
@@ -44,7 +72,7 @@ func (a *Archive) query(ctx context.Context, path, kql string) (queryTable, erro
}
request.Header.Set("Content-Type", "application/json")
request.Header.Set("x-ms-client-request-id", "cao-archive;"+uuid.NewString())
- response, err := a.queryHTTP.Do(request)
+ response, err := a.http.Do(request)
if err != nil {
return queryTable{}, ErrQuery
}
@@ -425,7 +453,7 @@ func (a *Archive) pinSource(ctx context.Context, scope sessionarchive.Scope, rep
if len(members) == 0 {
return nil, a.active.Generation, nil, nil, nil, sessionarchive.ErrNotFound
}
- if err := a.blob.private(ctx); err != nil {
+ if err := a.blob.CheckPrivate(ctx); err != nil {
return nil, 0, nil, nil, nil, err
}
// Persist the worst-case server execution horizon before issuing KQL.
diff --git a/server/internal/sessionarchiveblob/services.go b/server/internal/sessionarchiveblob/services.go
new file mode 100644
index 0000000000..4b6c1dc884
--- /dev/null
+++ b/server/internal/sessionarchiveblob/services.go
@@ -0,0 +1,54 @@
+package sessionarchiveblob
+
+import (
+ "context"
+ "errors"
+ "io"
+)
+
+var (
+ ErrObjectExists = errors.New("archive object already exists")
+ ErrNotFound = errors.New("archive object was not found")
+)
+
+// BlobStore owns one approved private object root. Immutable writes verify the
+// complete digest; pointer updates preserve ETag and lease preconditions.
+type BlobStore interface {
+ CheckPrivate(context.Context) error
+ Read(context.Context, string, int64) ([]byte, string, error)
+ WriteImmutable(context.Context, string, io.Reader, int64, string) error
+ Verify(context.Context, string, string, int64) error
+ CreatePointer(context.Context, []byte) error
+ ComparePointer(context.Context, []byte, string, string) (string, error)
+ Lease(context.Context, string, string) error
+ Delete(context.Context, string) error
+}
+
+// QueryService returns complete, bounded results from the approved ADX database.
+// Management supports existing metadata reads, not runtime provisioning.
+type QueryService interface {
+ Query(context.Context, string) (QueryResult, error)
+ Management(context.Context, string) (QueryResult, error)
+}
+
+type Services struct {
+ Blob BlobStore
+ Queries QueryService
+}
+
+func (s Services) validate() error {
+ if s.Blob == nil || s.Queries == nil {
+ return errors.New("hosted archive requires both Blob and ADX query services")
+ }
+ return nil
+}
+
+func azureServices(c Config) Services {
+ return Services{
+ Blob: &blobClient{root: c.blobRoot(), http: newHTTP(c, "https://storage.azure.com/.default")},
+ Queries: &adxQueryClient{config: c, http: newHTTP(c, "https://kusto.kusto.windows.net/.default")},
+ }
+}
+
+var _ BlobStore = (*blobClient)(nil)
+var _ QueryService = (*adxQueryClient)(nil)
diff --git a/server/internal/sessionarchiveblob/services_test.go b/server/internal/sessionarchiveblob/services_test.go
new file mode 100644
index 0000000000..0823c890a5
--- /dev/null
+++ b/server/internal/sessionarchiveblob/services_test.go
@@ -0,0 +1,144 @@
+package sessionarchiveblob
+
+import (
+ "context"
+ "errors"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/githubnext/gh-cao/server/internal/sessionarchive"
+ "github.com/githubnext/gh-cao/server/internal/simulator"
+)
+
+type failingBlobService struct {
+ BlobStore
+ failure error
+ calls int
+}
+
+func TestBlobDomainErrors(t *testing.T) {
+ for _, test := range []struct {
+ name string
+ status int
+ code string
+ exists bool
+ }{
+ {"conditional create", 412, "ConditionNotMet", true},
+ {"Azurite existing immutable blob", 409, "BlobAlreadyExists", true},
+ {"lease conflict is not an existing object", 409, "LeaseIdMismatchWithBlobOperation", false},
+ {"unknown conflict is not an existing object", 409, "", false},
+ {"denial is not an existing object", 403, "AuthorizationFailure", false},
+ } {
+ t.Run(test.name, func(t *testing.T) {
+ err := &blobResponseError{status: statusError(test.status), code: test.code}
+ if errors.Is(err, ErrObjectExists) != test.exists || !isStatus(err, test.status) {
+ t.Fatal("Blob error lost its domain meaning or original status")
+ }
+ })
+ }
+}
+
+func (s *failingBlobService) CheckPrivate(context.Context) error {
+ s.calls++
+ return s.failure
+}
+
+func TestArchiveServiceInjectionPreservesActivationGates(t *testing.T) {
+ capture := nativeCaptureConfig(t)
+ destination := validConfig()
+ destination.Acceptance = destination.NativeAcceptanceBinding()
+ failure := errors.New("synthetic Blob unavailable")
+ blob := &failingBlobService{failure: failure}
+ queries := &adxQueryClient{}
+ for _, test := range []struct {
+ name string
+ acceptance string
+ services Services
+ want error
+ }{
+ {"wrong acceptance before service access", "wrong", Services{Blob: blob, Queries: queries}, ErrConformance},
+ {"missing Blob service", destination.Acceptance, Services{Queries: queries}, nil},
+ {"missing ADX service", destination.Acceptance, Services{Blob: blob}, nil},
+ {"unavailable service without fallback", destination.Acceptance, Services{Blob: blob, Queries: queries}, failure},
+ } {
+ t.Run(test.name, func(t *testing.T) {
+ config := destination
+ config.Acceptance = test.acceptance
+ result, err := NewNativeWithServices(t.Context(), capture, config, test.services)
+ if err == nil || result != nil || test.want != nil && !errors.Is(err, test.want) {
+ t.Fatalf("expected explicit constructor failure: %v", err)
+ }
+ })
+ }
+ if blob.calls != 1 {
+ t.Fatalf("invalid activation reached Blob or failure retried a fallback: %d calls", blob.calls)
+ }
+ disabled, err := NewWithServices(t.Context(), Config{}, nil, Services{Blob: blob})
+ if err != nil || disabled.Status().Enabled || blob.calls != 1 {
+ t.Fatal("disabled archive attempted service access")
+ }
+}
+
+func TestArchiveInjectedServicesHTTPIntegration(t *testing.T) {
+ protocol := &blobProtocol{objects: map[string][]byte{}, etags: map[string]string{},
+ uploads: map[string]int{}, now: time.Now().UTC()}
+ server := httptest.NewServer(protocol)
+ t.Cleanup(server.Close)
+ transport, err := simulator.NewLocalTransport(server.URL)
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(transport.CloseIdleConnections)
+ client := &http.Client{Transport: transport, CheckRedirect: rejectRedirect, Timeout: 5 * time.Second}
+ config := validConfig()
+ queryConfig := config
+ queryConfig.Endpoint = server.URL
+ services := Services{
+ Blob: &blobClient{root: server.URL + "/container/archive", http: client},
+ Queries: &adxQueryClient{config: queryConfig, http: client},
+ }
+ config.Acceptance = config.NativeAcceptanceBinding()
+ archive, err := NewNativeWithServices(t.Context(), nativeCaptureConfig(t), config, services)
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() {
+ if err := archive.Close(); err != nil {
+ t.Error(err)
+ }
+ })
+ now := time.Now().UTC()
+ // A gap receipt needs no invented ADX event rows. Real content/engine
+ // acceptance remains separate from this injected-service recovery case.
+ if err := archive.CaptureSession(t.Context(), "octo/api", "123", 1, now, nil, "missing-session"); err == nil ||
+ !strings.Contains(err.Error(), "session archive acquisition gap: missing-session") {
+ t.Fatalf("gap admission must remain an explicit incomplete result: %v", err)
+ }
+ before := archive.Status()
+ revisions := archive.active.Revisions
+ if revisions == 0 || before.Gaps != 1 || before.ActiveTraces != 0 {
+ t.Fatalf("injected services did not persist the gap receipt: %+v", before)
+ }
+ if err := archive.Close(); err != nil {
+ t.Fatal(err)
+ }
+ reopened, err := NewNativeWithServices(t.Context(), nativeCaptureConfig(t), config, services)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer func() { _ = reopened.Close() }()
+ if reopened.Status().Generation != before.Generation || reopened.Status().Gaps != 1 || reopened.active.Revisions != revisions {
+ t.Fatal("restart did not recover publication through injected services")
+ }
+ if err := reopened.WithdrawRepository(t.Context(), "octo/api"); err != nil {
+ t.Fatal(err)
+ }
+ if reopened.Status().ActiveTraces != 0 || reopened.Status().Gaps != 0 {
+ t.Fatal("withdrawal did not persist through injected Blob service")
+ }
+}
+
+var _ sessionarchive.Publisher = (*Archive)(nil)
diff --git a/server/internal/tokenbroker/vault.go b/server/internal/tokenbroker/vault.go
index 21a0e57ac0..4d7ac8d577 100644
--- a/server/internal/tokenbroker/vault.go
+++ b/server/internal/tokenbroker/vault.go
@@ -27,8 +27,24 @@ type KeyVault struct {
SecretURL string
ClientID string
HTTPClient *http.Client
+ Identity ManagedIdentityTokenSource
}
+// ManagedIdentityTokenSource supplies a current token for the Key Vault audience.
+// Implementations must honor cancellation, reject stale tokens and sanitize errors.
+type ManagedIdentityTokenSource interface {
+ AccessToken(context.Context) (string, error)
+}
+
+// ManagedIdentity implements the App Service and IMDS credential protocols.
+type ManagedIdentity struct {
+ ClientID string
+ HTTPClient *http.Client
+}
+
+var _ PrivateKeyReader = KeyVault{}
+var _ ManagedIdentityTokenSource = ManagedIdentity{}
+
// PrivateKey reads the current secret value without caching credentials or keys.
func (v KeyVault) PrivateKey(ctx context.Context) ([]byte, error) {
secretURL, err := url.Parse(v.SecretURL)
@@ -50,10 +66,17 @@ func (v KeyVault) PrivateKey(ctx context.Context) ([]byte, error) {
}
client.Jar = nil
- token, err := v.managedIdentityToken(ctx, &client)
+ identity := v.Identity
+ if identity == nil {
+ identity = ManagedIdentity{ClientID: v.ClientID, HTTPClient: &client}
+ }
+ token, err := identity.AccessToken(ctx)
if err != nil {
return nil, err
}
+ if !validIdentityToken(token) {
+ return nil, errors.New("invalid managed identity response")
+ }
query := secretURL.Query()
query.Set("api-version", "7.4")
secretURL.RawQuery = query.Encode()
@@ -84,7 +107,18 @@ func safeVaultURL(u *url.URL) bool {
u.RawQuery == "" && !u.ForceQuery && u.Fragment == "" && u.RawFragment == ""
}
-func (v KeyVault) managedIdentityToken(ctx context.Context, client *http.Client) (string, error) {
+func (v ManagedIdentity) AccessToken(ctx context.Context) (string, error) {
+ client := http.Client{Timeout: 15 * time.Second}
+ if v.HTTPClient != nil {
+ client = *v.HTTPClient
+ if client.Timeout == 0 {
+ client.Timeout = 15 * time.Second
+ }
+ }
+ client.CheckRedirect = func(_ *http.Request, _ []*http.Request) error {
+ return http.ErrUseLastResponse
+ }
+ client.Jar = nil
endpoint, configured := os.LookupEnv("IDENTITY_ENDPOINT")
header := os.Getenv("IDENTITY_HEADER")
apiVersion := "2018-02-01"
@@ -117,7 +151,7 @@ func (v KeyVault) managedIdentityToken(ctx context.Context, client *http.Client)
} else {
request.Header.Set("Metadata", "true")
}
- body, err := vaultRequest(client, request)
+ body, err := vaultRequest(&client, request)
if err != nil {
return "", errors.New("managed identity request failed")
}
diff --git a/server/internal/tokenbroker/vault_integration_test.go b/server/internal/tokenbroker/vault_integration_test.go
new file mode 100644
index 0000000000..fbb7990c28
--- /dev/null
+++ b/server/internal/tokenbroker/vault_integration_test.go
@@ -0,0 +1,172 @@
+package tokenbroker
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/githubnext/gh-cao/server/internal/simulator"
+)
+
+// The injected transport maps only these fixture authorities to a real local
+// HTTP server. Production HTTPS validation and redirect rejection still run.
+func TestAzureKeyVaultHTTPIntegration(t *testing.T) {
+ for _, identity := range []string{"App Service", "IMDS"} {
+ t.Run(identity, func(t *testing.T) {
+ unsetVaultIdentityEnvironment(t)
+ const secretURL = "https://example.vault.azure.net/secrets/signing-key" // #nosec G101 -- synthetic secret reference, not a credential.
+ const appEndpoint = "http://127.0.0.1:8081/identity/token"
+ const imdsEndpoint = "http://169.254.169.254/metadata/identity/oauth2/token"
+ endpoint := imdsEndpoint
+ if identity == "App Service" {
+ endpoint = appEndpoint
+ t.Setenv("IDENTITY_ENDPOINT", endpoint)
+ t.Setenv("IDENTITY_HEADER", "synthetic-deployment-header")
+ }
+ var mu sync.Mutex
+ value, identityStatus, secretStatus := "synthetic-key-v1", http.StatusOK, http.StatusOK
+ expiry := time.Now().Add(time.Hour).Unix()
+ identityReads, secretReads := 0, 0
+ local := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ mu.Lock()
+ defer mu.Unlock()
+ if r.Method != http.MethodGet {
+ http.Error(w, "unsupported method", http.StatusMethodNotAllowed)
+ return
+ }
+ w.Header().Set("Content-Type", "application/json")
+ if r.Host == "example.vault.azure.net" && r.URL.Path == "/secrets/signing-key" {
+ secretReads++
+ if r.Header.Get("Authorization") != "Bearer synthetic-identity-token" ||
+ r.Header.Get("X-IDENTITY-HEADER") != "" || r.Header.Get("Metadata") != "" ||
+ r.URL.RawQuery != "api-version=7.4" {
+ t.Error("secret request violated the vault contract")
+ w.WriteHeader(http.StatusBadRequest)
+ return
+ }
+ w.Header().Set("Location", "https://untrusted.invalid/secret")
+ w.WriteHeader(secretStatus)
+ _ = json.NewEncoder(w).Encode(map[string]string{"value": value})
+ return
+ }
+ if r.Host+r.URL.Path != strings.TrimPrefix(endpoint, "http://") {
+ t.Error("unexpected identity authority or path")
+ w.WriteHeader(http.StatusNotFound)
+ return
+ }
+ identityReads++
+ version := "2018-02-01"
+ if identity == "App Service" {
+ version = "2019-08-01"
+ if r.Header.Get("X-IDENTITY-HEADER") != "synthetic-deployment-header" {
+ t.Error("missing App Service identity header")
+ }
+ } else if r.Header.Get("Metadata") != "true" {
+ t.Error("missing IMDS metadata header")
+ }
+ if r.Header.Get("Authorization") != "" || r.URL.Query().Get("resource") != "https://vault.azure.net" ||
+ r.URL.Query().Get("client_id") != "synthetic-client" || r.URL.Query().Get("api-version") != version {
+ t.Error("identity request violated its audience/client contract")
+ }
+ w.Header().Set("Location", "https://untrusted.invalid/identity")
+ w.WriteHeader(identityStatus)
+ _ = json.NewEncoder(w).Encode(map[string]any{
+ "access_token": "synthetic-identity-token", "token_type": "Bearer", "expires_on": expiry,
+ })
+ }))
+ t.Cleanup(local.Close)
+ transport, err := simulator.NewLocalTransport(local.URL)
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(transport.CloseIdleConnections)
+ client := &http.Client{Transport: vaultTestTransport(func(r *http.Request) (*http.Response, error) {
+ authority := r.URL.Scheme + "://" + r.URL.Host + r.URL.Path
+ if authority != endpoint && authority != secretURL {
+ return nil, errors.New("fixture blocked unknown authority")
+ }
+ mapped := r.Clone(r.Context())
+ url := *r.URL
+ mapped.URL = &url
+ mapped.URL.Scheme = "http"
+ mapped.URL.Host = strings.TrimPrefix(local.URL, "http://")
+ mapped.Host = r.URL.Host
+ return transport.RoundTrip(mapped)
+ })}
+ var identitySource ManagedIdentityTokenSource = ManagedIdentity{ClientID: "synthetic-client", HTTPClient: client}
+ var vault PrivateKeyReader = KeyVault{SecretURL: secretURL, HTTPClient: client, Identity: identitySource}
+ for _, version := range []string{"synthetic-key-v1", "synthetic-key-v2"} {
+ mu.Lock()
+ value = version
+ mu.Unlock()
+ key, err := vault.PrivateKey(t.Context())
+ if err != nil || string(key) != version {
+ t.Fatalf("fresh secret read failed: %v", err)
+ }
+ }
+ mu.Lock()
+ if identityReads != 2 || secretReads != 2 {
+ t.Error("rotation must reread both identity and secret")
+ }
+ mu.Unlock()
+ for _, failure := range []struct {
+ name string
+ identityStatus, status int
+ expired bool
+ empty bool
+ secretRequests int
+ }{
+ {"identity denied", 403, 200, false, false, 0},
+ {"identity unavailable", 503, 200, false, false, 0},
+ {"identity redirect", 307, 200, false, false, 0},
+ {"expired identity", 200, 200, true, false, 0},
+ {"secret denied", 200, 403, false, false, 1},
+ {"secret missing", 200, 404, false, false, 1},
+ {"secret unavailable", 200, 503, false, false, 1},
+ {"secret redirect", 200, 307, false, false, 1},
+ {"empty secret", 200, 200, false, true, 1},
+ } {
+ t.Run(failure.name, func(t *testing.T) {
+ mu.Lock()
+ identityStatus, secretStatus = failure.identityStatus, failure.status
+ value, expiry = "synthetic-sensitive-upstream", time.Now().Add(time.Hour).Unix()
+ if failure.expired {
+ expiry = 1
+ }
+ if failure.empty {
+ value = ""
+ }
+ beforeIdentity, beforeSecret := identityReads, secretReads
+ mu.Unlock()
+ key, err := vault.PrivateKey(t.Context())
+ if err == nil || key != nil || strings.Contains(err.Error(), "synthetic-") {
+ t.Fatal("upstream failure must return a sanitized error without a key")
+ }
+ mu.Lock()
+ defer mu.Unlock()
+ if identityReads != beforeIdentity+1 || secretReads != beforeSecret+failure.secretRequests {
+ t.Error("failure followed a redirect or continued after identity rejection")
+ }
+ })
+ }
+ if client.Timeout != 0 || client.CheckRedirect != nil {
+ t.Fatal("KeyVault mutated the injected client")
+ }
+ t.Run("cancelled read", func(t *testing.T) {
+ ctx := t.Context()
+ // An already expired context must not make a network request.
+ ctx, cancel := context.WithDeadline(ctx, time.Now().Add(-time.Second))
+ defer cancel()
+ if key, err := vault.PrivateKey(ctx); err == nil || key != nil {
+ t.Fatalf("cancelled read returned a key: %v", err)
+ }
+ })
+ })
+ }
+}
diff --git a/specs/session-archive.md b/specs/session-archive.md
index 88775def6e..e69bd1d0f9 100644
--- a/specs/session-archive.md
+++ b/specs/session-archive.md
@@ -132,6 +132,17 @@ Unsupported filesystems or insufficient permissions MUST fail closed. The
legacy hosted profile additionally requires a persistent private staging volume, one
Blob lease owner, and conditional active-pointer updates.
+Direct Blob and ADX access MUST cross the owned `BlobStore` and `QueryService`
+interfaces. Consumers MUST NOT receive raw HTTP responses or concrete provider
+clients. The composition root selects the production adapters; explicitly
+injected services use the same activation, destination, scope, ownership,
+bounded-result and recovery contracts. `NewWithServices` and
+`NewNativeWithServices` MUST reject missing services and MUST NOT substitute a
+different backend after an adapter failure. Blob contracts preserve immutable
+content verification, conditional ETag/lease pointer publication and explicit
+not-found/conflict errors. Query contracts return complete bounded results or
+an error; protocol fixtures cannot establish real Azure conformance.
+
The explicit native hosted profile instead owns one Blob lease and conditional
active-pointer updates, without a POSIX staging volume. It does not relax the
single-exporter, OAuth, persistent-host lifecycle or real-destination acceptance
diff --git a/tests/integration/azure-local-services.test.mjs b/tests/integration/azure-local-services.test.mjs
new file mode 100644
index 0000000000..245606eabc
--- /dev/null
+++ b/tests/integration/azure-local-services.test.mjs
@@ -0,0 +1,238 @@
+import assert from "node:assert/strict";
+import { execFile } from "node:child_process";
+import { createHmac, randomBytes } from "node:crypto";
+import { readFile } from "node:fs/promises";
+import http from "node:http";
+import { fileURLToPath } from "node:url";
+import { promisify } from "node:util";
+import test from "node:test";
+
+const runtimePath = process.argv[2];
+if (!runtimePath) throw new Error("runtime state JSON path is required");
+const runtime = JSON.parse(await readFile(runtimePath, "utf8"));
+const root = fileURLToPath(new URL("../../", import.meta.url));
+const execute = promisify(execFile);
+
+function localEndpoint(value, protocol) {
+ const url = new URL(value);
+ assert.equal(url.protocol, protocol);
+ assert.equal(url.hostname, "127.0.0.1", "service tests require numeric loopback");
+ assert.ok(url.port, "service tests require an explicit port");
+ assert.equal(url.username, "");
+ assert.equal(url.password, "");
+ assert.equal(url.search, "");
+ assert.equal(url.hash, "");
+ return url;
+}
+
+function storageRequest(service, method, path, body = "", headers = {}, invalidKey = false) {
+ const endpoint = localEndpoint(runtime.storage[`${service}Endpoint`], "http:");
+ assert.match(runtime.storage.account, /^[a-z0-9]+$/);
+ assert.equal(endpoint.pathname, `/${runtime.storage.account}`);
+ const url = new URL(`${endpoint}${path}`);
+ const date = new Date().toUTCString();
+ const content = Buffer.from(body);
+ headers = {
+ "x-ms-date": date,
+ "x-ms-version": "2019-02-02",
+ "content-length": String(content.length),
+ ...headers,
+ };
+ const resource = `/${runtime.storage.account}${url.pathname}`;
+ let canonical;
+ if (service === "table") {
+ canonical = `${date}\n${resource}`;
+ } else {
+ const standard = [
+ method, headers["content-encoding"], headers["content-language"],
+ content.length ? String(content.length) : "", headers["content-md5"],
+ headers["content-type"], "", headers["if-modified-since"], headers["if-match"],
+ headers["if-none-match"], headers["if-unmodified-since"], headers.range,
+ ].map((value) => value || "").join("\n");
+ const azureHeaders = Object.keys(headers).filter((name) => name.startsWith("x-ms-"))
+ .sort().map((name) => `${name}:${headers[name]}\n`).join("");
+ const query = [...new Set([...url.searchParams.keys()].map((name) => name.toLowerCase()))]
+ .sort().map((name) => `\n${name}:${url.searchParams.getAll(name).sort().join(",")}`).join("");
+ canonical = `${standard}\n${azureHeaders}${resource}${query}`;
+ }
+ const key = invalidKey ? randomBytes(32) : Buffer.from(runtime.storage.key, "base64");
+ const signature = createHmac("sha256", key).update(canonical).digest("base64");
+ headers.authorization = `${service === "table" ? "SharedKeyLite" : "SharedKey"} ${runtime.storage.account}:${signature}`;
+ // Node's HTTP client does not consult ambient proxy settings or follow redirects.
+ return new Promise((resolve, reject) => {
+ const request = http.request(url, { method, headers, agent: false }, (response) => {
+ const chunks = [];
+ let length = 0;
+ response.on("data", (chunk) => {
+ length += chunk.length;
+ if (length > 1 << 20) response.destroy(new Error("emulator response exceeds test limit"));
+ else chunks.push(chunk);
+ });
+ response.on("error", reject);
+ response.on("end", () => resolve({
+ status: response.statusCode,
+ headers: response.headers,
+ body: Buffer.concat(chunks).toString("utf8"),
+ }));
+ });
+ request.setTimeout(5000, () => request.destroy(new Error("emulator request timed out")));
+ request.on("error", reject);
+ request.end(content);
+ });
+}
+
+test("Azure Storage Blob: signed writes, reads, ETag conflicts and denied credentials", async (t) => {
+ const container = `cao-test-${randomBytes(8).toString("hex")}`;
+ assert.equal((await storageRequest("blob", "PUT", `/${container}?restype=container`)).status, 201);
+ t.after(async () => {
+ assert.equal((await storageRequest("blob", "DELETE", `/${container}?restype=container`)).status, 202);
+ });
+ const path = `/${container}/evidence.json`;
+ const written = await storageRequest("blob", "PUT", path, '{"synthetic":true}', {
+ "x-ms-blob-type": "BlockBlob",
+ });
+ assert.equal(written.status, 201);
+ const read = await storageRequest("blob", "GET", path);
+ assert.equal(read.status, 200);
+ assert.deepEqual(JSON.parse(read.body), { synthetic: true });
+ assert.equal(read.headers.etag, written.headers.etag);
+ assert.equal((await storageRequest("blob", "PUT", path, "replacement", {
+ "x-ms-blob-type": "BlockBlob", "if-match": '"stale"',
+ })).status, 412);
+ assert.equal((await storageRequest("blob", "GET", path, "", {}, true)).status, 403);
+ assert.equal((await storageRequest("blob", "GET", path)).body, read.body);
+ assert.equal((await storageRequest("blob", "GET", `/${container}/missing`)).status, 404);
+});
+
+test("Azure Storage Queue: visibility, receipts and acknowledgement", async (t) => {
+ const queue = `cao-test-${randomBytes(8).toString("hex")}`;
+ assert.equal((await storageRequest("queue", "PUT", `/${queue}`)).status, 201);
+ t.after(async () => {
+ assert.equal((await storageRequest("queue", "DELETE", `/${queue}`)).status, 204);
+ });
+ const messages = `/${queue}/messages`;
+ assert.equal((await storageRequest("queue", "POST", messages,
+ "synthetic-message",
+ { "content-type": "application/xml" })).status, 201);
+ const received = await storageRequest("queue", "GET", `${messages}?visibilitytimeout=30`);
+ assert.equal(received.status, 200);
+ assert.match(received.body, /synthetic-message<\/MessageText>/);
+ const id = received.body.match(/([^<]+)<\/MessageId>/)?.[1];
+ const receipt = received.body.match(/([^<]+)<\/PopReceipt>/)?.[1];
+ assert.ok(id);
+ assert.ok(receipt);
+ const hidden = await storageRequest("queue", "GET", messages);
+ assert.equal(hidden.status, 200);
+ assert.doesNotMatch(hidden.body, //);
+ assert.equal((await storageRequest("queue", "DELETE",
+ `${messages}/${encodeURIComponent(id)}?popreceipt=${encodeURIComponent(receipt)}`)).status, 204);
+ const empty = await storageRequest("queue", "GET", messages);
+ assert.equal(empty.status, 200);
+ assert.doesNotMatch(empty.body, //);
+});
+
+test("Azure Storage Table: entity persistence and stale ETag rejection", async (t) => {
+ const table = `CaoTest${randomBytes(8).toString("hex")}`;
+ const headers = { "content-type": "application/json", accept: "application/json;odata=nometadata" };
+ assert.equal((await storageRequest("table", "POST", "/Tables",
+ JSON.stringify({ TableName: table }), headers)).status, 201);
+ t.after(async () => {
+ assert.equal((await storageRequest("table", "DELETE", `/Tables('${table}')`, "", headers)).status, 204);
+ });
+ const entity = { PartitionKey: "synthetic", RowKey: "one", value: "original" };
+ const inserted = await storageRequest("table", "POST", `/${table}`, JSON.stringify(entity), headers);
+ assert.equal(inserted.status, 201);
+ const path = `/${table}(PartitionKey='synthetic',RowKey='one')`;
+ const read = await storageRequest("table", "GET", path, "", headers);
+ assert.equal(read.status, 200);
+ assert.equal(JSON.parse(read.body).value, "original");
+ assert.ok(read.headers.etag);
+ assert.equal((await storageRequest("table", "PUT", path,
+ JSON.stringify({ ...entity, value: "updated" }), { ...headers, "if-match": read.headers.etag })).status, 204);
+ assert.equal((await storageRequest("table", "PUT", path,
+ JSON.stringify({ ...entity, value: "stale" }), { ...headers, "if-match": read.headers.etag })).status, 412);
+ assert.equal(JSON.parse((await storageRequest("table", "GET", path, "", headers)).body).value, "updated");
+ assert.equal((await storageRequest("table", "GET",
+ `/${table}(PartitionKey='synthetic',RowKey='missing')`, "", headers)).status, 404);
+});
+
+async function goCase(directory, packages, pattern, services = {}) {
+ const env = Object.fromEntries([
+ "PATH", "HOME", "TMPDIR", "GOPATH", "GOMODCACHE", "GOCACHE", "SDKROOT", "DEVELOPER_DIR",
+ ].filter((name) => process.env[name] !== undefined).map((name) => [name, process.env[name]]));
+ Object.assign(env, {
+ GOTOOLCHAIN: "local", GOPROXY: "off", GOSUMDB: "off", GOVCS: "*:off",
+ GOENV: "off", GOFLAGS: "-mod=readonly", DO_NOT_TRACK: "1", GH_TELEMETRY: "0",
+ OTEL_SDK_DISABLED: "true",
+ });
+ if (services.postgres) {
+ const url = new URL(runtime.postgresUrl);
+ assert.equal(url.protocol, "postgres:");
+ assert.equal(url.hostname, "127.0.0.1");
+ assert.ok(url.port);
+ assert.equal(url.search, "?sslmode=disable", "connection overrides are forbidden");
+ env.POSTGRES_URL = url.href;
+ }
+ if (services.blob) {
+ localEndpoint(runtime.storage.blobEndpoint, "http:");
+ Object.assign(env, {
+ CAO_AZURITE_BLOB_URL: runtime.storage.blobEndpoint,
+ CAO_AZURITE_ACCOUNT: runtime.storage.account,
+ CAO_AZURITE_KEY: runtime.storage.key,
+ });
+ }
+ const { stdout } = await execute("go", [
+ "test", "-race", ...packages, "-run", pattern, "-count=1", "-timeout=3m", "-json",
+ ], { cwd: `${root}${directory}`, env, timeout: 240_000, maxBuffer: 8 << 20 });
+ const events = stdout.trim().split("\n").map((line) => JSON.parse(line));
+ assert.ok(events.some((event) => event.Test && event.Action === "pass"), "selector must execute tests");
+ assert.deepEqual(events.filter((event) => event.Action === "skip"), [], "selected service coverage cannot skip");
+}
+
+test("Azure PostgreSQL canonical adapter: publication, reopen and contract drift", async () => {
+ await goCase("server", ["./internal/postgresx"], "^TestFreshSchemaReopen$", { postgres: true });
+});
+
+test("Azure PostgreSQL operational adapter: sessions, CAS, queue capacity and leases", async () => {
+ await goCase("server", ["./internal/operational"], "^TestAdapters$/^postgres$", { postgres: true });
+});
+
+test("Azure Blob archive adapter: real Azurite immutable objects, leases and pointer CAS", async () => {
+ await goCase("server", ["./internal/sessionarchiveblob"], "^TestAzuriteBlobStoreIntegration$", { blob: true });
+});
+
+test("Azure archive service interfaces: injection, activation gates and HTTP recovery", async () => {
+ await goCase("server", ["./internal/sessionarchiveblob"], "^TestArchive(ServiceInjectionPreservesActivationGates|InjectedServicesHTTPIntegration)$");
+});
+
+test("Azure Key Vault and managed identity: local HTTP protocol integration", async () => {
+ await goCase("server", ["./internal/tokenbroker"], "^TestAzureKeyVaultHTTPIntegration$");
+});
+
+test("Azure monitoring: template-driven OTLP receipts and exporter failure isolation", async () => {
+ await goCase("server", ["./internal/telemetry"], "^TestAzureDeploymentTelemetryReceipts$");
+});
+
+test("Azure Container Registry protocol: digest bindings and configured-image read-back", async () => {
+ await goCase("", ["./internal/cli"],
+ "^TestDeployAzure(ImageBindingsFailClosed|ImageReadBackBlocksJobsOnMismatch)$");
+});
+
+test("Azure Container Apps jobs protocol: bounded cancellation", async () => {
+ await goCase("", ["./internal/cli"], "^TestDeployAzureJobWaitCancellation$");
+});
+
+test("Azure Functions configuration protocol: unresolved Key Vault references fail closed", async () => {
+ await goCase("", ["./internal/cli"], "^TestDeployAzureApplyFailsClosedWhenKeyVaultReferencesStayUnresolved$");
+});
+
+test("Azure Resource Manager protocol: read-only planning and required preflight checks", async () => {
+ await goCase("", ["./internal/cli"], "^TestDeployAzure(PlanRunsOnlyReadOnlyCommands|PreflightChecks|PreflightHealthy)$");
+});
+
+test("Azure networking, RBAC and optional ADX: exhaustive Bicep contract coverage", async () => {
+ await execute(process.execPath, [
+ "--test", "tests/unit/server-azure-service-coverage.test.mjs",
+ "tests/unit/server-azure-bicep.test.mjs", "tests/unit/server-azure-adx.test.mjs",
+ ], { cwd: root, timeout: 30_000, maxBuffer: 4 << 20 });
+});
diff --git a/tests/unit/server-azure-service-coverage.test.mjs b/tests/unit/server-azure-service-coverage.test.mjs
new file mode 100644
index 0000000000..3b96bbf138
--- /dev/null
+++ b/tests/unit/server-azure-service-coverage.test.mjs
@@ -0,0 +1,62 @@
+import assert from "node:assert/strict";
+import { readdir, readFile } from "node:fs/promises";
+import test from "node:test";
+
+// This is an inventory guard, not evidence that ARM or managed services ran.
+const coverage = {
+ "Microsoft.Web/sites": "Functions Core Tools",
+ "Microsoft.Web/serverfarms": "deployment contract only",
+ "Microsoft.Storage/storageAccounts": "Azurite Blob, Queue and Table",
+ "Microsoft.DBforPostgreSQL/flexibleServers": "PostgreSQL 16",
+ "Microsoft.DBforPostgreSQL/flexibleServers/databases": "PostgreSQL 16",
+ "Microsoft.ContainerRegistry/registries": "Azure CLI protocol simulator",
+ "Microsoft.App/managedEnvironments": "deployment contract only",
+ "Microsoft.App/containerApps": "deployment contract only",
+ "Microsoft.App/jobs": "Azure CLI job protocol simulator",
+ "Microsoft.KeyVault/vaults": "local identity/secret protocol fixture",
+ "Microsoft.KeyVault/vaults/secrets": "local identity/secret protocol fixture",
+ "Microsoft.ManagedIdentity/userAssignedIdentities": "local identity protocol fixture",
+ "Microsoft.Authorization/roleAssignments": "deployment contract only",
+ "Microsoft.OperationalInsights/workspaces": "deployment contract only",
+ "Microsoft.Insights/components": "OTLP protocol receiver, not Application Insights",
+ "Microsoft.Insights/diagnosticSettings": "deployment contract only",
+ "Microsoft.Network/virtualNetworks": "deployment contract only",
+ "Microsoft.Network/privateDnsZones": "deployment contract only",
+ "Microsoft.Network/privateDnsZones/virtualNetworkLinks": "deployment contract only",
+ "Microsoft.Network/privateEndpoints": "deployment contract only",
+ "Microsoft.Network/privateEndpoints/privateDnsZoneGroups": "deployment contract only",
+ "Microsoft.Kusto/clusters": "optional official Kusto engine",
+ "Microsoft.Kusto/clusters/databases": "optional official Kusto engine",
+};
+const directory = new URL("../../server/azure/", import.meta.url);
+const bicepTypes = new Set();
+for (const name of await readdir(directory)) {
+ if (!name.endsWith(".bicep")) continue;
+ const source = await readFile(new URL(name, directory), "utf8");
+ for (const match of source.matchAll(/^\s*resource\s+\w+\s+'([^'@]+)@[^']+'/gm)) {
+ bicepTypes.add(match[1]);
+ }
+}
+const generated = JSON.parse(await readFile(new URL("main.json", directory), "utf8"));
+const compiledTypes = new Set();
+function visit(value) {
+ if (!value || typeof value !== "object") return;
+ if (typeof value.type === "string" && value.type.startsWith("Microsoft.")
+ && value.type !== "Microsoft.Resources/deployments") compiledTypes.add(value.type);
+ for (const child of Object.values(value)) visit(child);
+}
+visit(generated);
+
+test("Every Bicep service has an explicit local mechanism or unavailable managed-service boundary", () => {
+ assert.deepEqual([...bicepTypes].sort(), Object.keys(coverage).sort(),
+ "new or removed Bicep resource types require a local-coverage review");
+ assert.deepEqual([...compiledTypes].sort(), [...bicepTypes].sort(),
+ "compiled resources must match the reviewed Bicep service inventory");
+});
+
+const documentation = await readFile(new URL("../../scripts/azure-local/README.md", import.meta.url), "utf8");
+for (const [type, mechanism] of Object.entries(coverage)) {
+ test(`${type}: ${mechanism}`, () => {
+ assert.ok(documentation.includes(`\`${type}\``), "document the service's local coverage and limitations");
+ });
+}