stroblmeandClaude Opus 5
4c8339e643
A screen somewhere this installation is not reachable from asks the portal for a code instead, and the portal mints its credential — because a token signed here is one such a device could never present. Where it was minted changes nothing about what it may do. The panel gate moved off the branch that decodes a local panel token and onto whatever claims name a panel, so the portal's and this installation's are bounded by the same check against the same panel's dashboards. A token of that scope naming no panel is refused rather than left holding the account it borrows. The connector marks what arrives on its socket, since that is the only thing that makes it true, and the approval screen now names what is holding a code — approving adopts whatever answers, so it is worth a look first. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017F9RnYCJgASuBTcAjxmnsp
113 lines
3.7 KiB
Python
113 lines
3.7 KiB
Python
import uuid
|
|
from collections.abc import Generator
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine, text
|
|
from sqlalchemy.engine import make_url
|
|
from sqlmodel import Session, SQLModel, select
|
|
|
|
from app.cloud import config as cloud_config
|
|
from app.core.config import settings
|
|
from app.core.db import engine, init_db
|
|
from app.main import app
|
|
from app.models import User
|
|
from tests.utils.portal import INSTALLATION_ID, ISSUER, jwks
|
|
from tests.utils.user import authentication_token_from_email
|
|
from tests.utils.utils import get_superuser_token_headers
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def flow_data(tmp_path_factory: pytest.TempPathFactory) -> Generator[None, None, None]:
|
|
"""Keep flows and secrets written by tests out of the real store."""
|
|
root = tmp_path_factory.mktemp("flow-data")
|
|
settings.FLOWS_DIR = root / "flows"
|
|
settings.SECRETS_FILE = root / "secrets.enc"
|
|
settings.PANELS_FILE = root / "panels.json"
|
|
yield
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def db() -> Generator[Session, None, None]:
|
|
"""Create the throwaway database `tests/__init__.py` points at, drop it after."""
|
|
url = make_url(str(settings.SQLALCHEMY_DATABASE_URI))
|
|
# The teardown drops this database, so refuse to run against anything but
|
|
# the dedicated test one.
|
|
assert url.database and url.database.endswith("_test"), url.database
|
|
|
|
maintenance = create_engine(
|
|
url.set(database="postgres"), isolation_level="AUTOCOMMIT"
|
|
)
|
|
drop = text(f'DROP DATABASE IF EXISTS "{url.database}" WITH (FORCE)')
|
|
with maintenance.connect() as connection:
|
|
connection.execute(drop)
|
|
connection.execute(text(f'CREATE DATABASE "{url.database}"'))
|
|
|
|
SQLModel.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
init_db(session)
|
|
yield session
|
|
|
|
engine.dispose()
|
|
with maintenance.connect() as connection:
|
|
connection.execute(drop)
|
|
maintenance.dispose()
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def client() -> Generator[TestClient, None, None]:
|
|
with TestClient(app) as c:
|
|
yield c
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def superuser_token_headers(client: TestClient) -> dict[str, str]:
|
|
return get_superuser_token_headers(client)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def normal_user_token_headers(client: TestClient, db: Session) -> dict[str, str]:
|
|
return authentication_token_from_email(
|
|
client=client, email=settings.EMAIL_TEST_USER, db=db
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def portal_key() -> rsa.RSAPrivateKey:
|
|
return rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
|
|
|
|
@pytest.fixture
|
|
def enrolled(
|
|
tmp_path_factory: pytest.TempPathFactory,
|
|
portal_key: rsa.RSAPrivateKey,
|
|
db: Session,
|
|
) -> Any:
|
|
"""Enrol this installation with a fake portal, then undo it."""
|
|
local_user = db.exec(
|
|
select(User).where(User.email == settings.FIRST_SUPERUSER)
|
|
).one()
|
|
original = settings.CLOUD_CONFIG_FILE
|
|
settings.CLOUD_CONFIG_FILE = (
|
|
tmp_path_factory.mktemp(f"cloud-{uuid.uuid4().hex[:6]}") / "cloud.json"
|
|
)
|
|
cloud_config.save(
|
|
cloud_config.CloudConfig(
|
|
portal_url=ISSUER,
|
|
ws_url=f"{ISSUER}/api/v1/tunnel/attach",
|
|
installation_id=INSTALLATION_ID,
|
|
token="installation-token",
|
|
issuer=ISSUER,
|
|
jwks=jwks(portal_key),
|
|
local_user_id=str(local_user.id),
|
|
enrolled_at=datetime.now(UTC).isoformat(),
|
|
portal_account=settings.FIRST_SUPERUSER,
|
|
)
|
|
)
|
|
yield local_user
|
|
cloud_config.delete()
|
|
settings.CLOUD_CONFIG_FILE = original
|