Metadata-Version: 2.4
Name: scalio-connect
Version: 0.3.0
Summary: Signed calls between the Scalio apps and Compass (format v2)
License: Proprietary
Requires-Python: >=3.11
Description-Content-Type: text/markdown
Requires-Dist: cryptography>=42
Provides-Extra: web
Requires-Dist: pymongo>=4.6; extra == "web"
Provides-Extra: support
Requires-Dist: httpx>=0.27; extra == "support"
Requires-Dist: pymongo>=4.6; extra == "support"
Provides-Extra: dev
Requires-Dist: pytest>=8; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23; extra == "dev"
Requires-Dist: pymongo>=4.6; extra == "dev"
Requires-Dist: mongomock-motor>=0.0.29; extra == "dev"
Requires-Dist: typing_extensions>=4.9; extra == "dev"
Requires-Dist: ruff>=0.6; extra == "dev"
Requires-Dist: httpx>=0.27; extra == "dev"

# scalio-connect

Signed calls between the Scalio apps and Compass. One format ("v2", header `Scalio-Signature`) for every
direction:

- Compass -> app: control key K3 on `/api/provision/*` (replaces the old `X-Provision-Control-Signature`)
- app -> Compass: instance key on the support intake `/api/support/v1/*` (`scalio_connect.support`)

Contract: scalio-docs `interfaces/signed-request-v2.md`. Decision: scalio-docs ADR 0020.

## What is signed

```
Scalio-Signature: v=2,kid=k3,ts=1793131200,nonce=vector-nonce-0000000001,sig=<base64 Ed25519>

scalio-sig-v2
<kid>
<METHOD>
<audience = receiver origin, e.g. https://acme.eu.scalionexus.com>
<path and query as sent, e.g. /api/provision/admin>
<ts>
<nonce>
<sha256 of the raw body, hex>
```

Lines joined by `\n`, no line break at the end. Time window +-300 s, each (kid, nonce) once.
Because receiver and path are signed, a captured request does not work on another instance or another
endpoint (security review H9).

## Usage

```python
from scalio_connect import sign, verify, SignatureError
from scalio_connect.nonces import use_nonce

# sender
headers = sign(private_key, "k3", "POST", "https://acme.eu.scalionexus.com/api/provision/admin", body)

# receiver (raw body bytes, path with query as received)
try:
    v = verify(request.headers["Scalio-Signature"], request.method, path_with_query, body,
               audiences=["https://acme.eu.scalionexus.com"], keys_for=lambda kid: k3_keys if kid == "k3" else [])
except SignatureError:
    raise HTTPException(401)
if not await use_nonce(db, v.kid, v.nonce):
    raise HTTPException(401)
```

`audiences` are the receiver's own public origins from its configuration. Do not take them from the
`Host` header.

## K3 during the switch

```python
from scalio_connect.k3 import K3Error, audiences_from_env, check_request

try:
    await check_request(db, request.headers, request.method, path_with_query, body,
                        keys=k3_public_keys, audiences=audiences_from_env())
except K3Error as e:
    raise HTTPException(401, str(e))
```

Accepts v2 and the old `X-Provision-*` format until the first valid v2 call, then only v2. Before that a
failed v2 check falls back to the old format (logged), so a wrong public URL does not lock Compass out.
Public URLs come from `SCALIO_PUBLIC_URLS`, `APP_PUBLIC_URL`, `FRONTEND_URL`, `PUBLIC_URL`.

## Development

```
pip install -e ".[dev]"
ruff check . && pytest -q
```

`tests/vector.json` is the shared test vector (throwaway seed). Compass and every app test against the same file.
Release: bump `pyproject.toml`, CHANGELOG line, merge, then "Publish" (cockpit or Actions) with the version, no `v`.

## Support connection (app -> Compass)

`pip install "scalio-connect[support]"`. Decision: scalio-docs ADR 0022, contract `interfaces/support-intake.md`.

```python
from scalio_connect.support import SupportClient, cipher_from_secret, build_diagnostics
from scalio_connect.support import connection, outbox

cipher = cipher_from_secret(os.environ["JWT_SECRET"])          # any app secret of 16+ characters
conn = await connection.load(db, cipher) or await connection.create(db, cipher, "nexus", portal_url)
client = SupportClient(conn)
await client.pair(app="nexus", version=VERSION, mode="saas", code=one_time_code)   # SaaS / dev / test
await connection.set_status(db, "active")

item = await outbox.enqueue(db, "ticket", {"subject": ..., "description": ..., "reporter": {...},
                                           "diagnostics": build_diagnostics(app="nexus", ...)},
                            reporter_email=user_email)
await outbox.flush(db, client if conn.usable else None)       # again when the drawer opens
```

The instance key never leaves the instance. If the app secret changes, the key cannot be decrypted any more and
the admin connects again.
