fix(r2): upgrade on any chart change, use FM's internal port, mask tokens in test output

oop_platform_partners: []

oop_platform_fm_deployment: federation-manager

# Federation Manager's internal API as seen from inside its own pod.

oop_platform_fm_internal_url: http://127.0.0.1:8082

oop_platform_fm_internal_url: http://127.0.0.1:8083

      {{ oop_platform_chart_path if oop_platform_chart_path | length > 0

         else oop_platform_work_dir ~ '/helm/oop-platform-chart' }}

# The helm module upgrades a release only when the chart version or the values change,

# and the chart keeps one version across many changes. Its files' digest, passed as a

# value, makes any change to them an upgrade. Dependency archives are left out: Chart.yaml

# pins their versions, and `helm dependency update` rewrites them on every run.

- name: Fingerprint the chart

  ansible.builtin.shell: >-

    set -o pipefail;

    find . -type f ! -path './charts/*.tgz' ! -path '*/.git/*' -print0

    | LC_ALL=C sort -z | xargs -0 sha256sum | sha256sum | cut -d' ' -f1

  args:

    chdir: "{{ oop_platform_chart }}"

    executable: /bin/bash

  register: oop_platform_chart_digest

  changed_when: false

- name: Build the chart values

  ansible.builtin.set_fact:

    oop_platform_values: "{{ oop_platform_base_values | combine(oop_platform_optional_values, oop_platform_extra_values, recursive=true) }}"

  vars:

    oop_platform_base_values:

      chartDigest: "{{ oop_platform_chart_digest.stdout }}"

      global:

        namespace: "{{ oop_platform_namespace }}"

        publicHost: "{{ oop_platform_public_host }}"

"""

class Token(str):

    """An access token that reads as <token> wherever pytest prints it, such as the

    fixture values in a failed test's traceback."""

    def __repr__(self) -> str:

        return "<token>"

@dataclass

class Operator:

    host: str

    def secret(self, name: str, key: str) -> str:

        if (name, key) not in self._secrets:

            data = json.loads(self.kubectl("get", "secret", name, "-o", "json"))["data"]

            self._secrets[(name, key)] = base64.b64decode(data[key]).decode()

            self._secrets[(name, key)] = Token(base64.b64decode(data[key]).decode())

        return self._secrets[(name, key)]

    def fm_internal(self, method: str, path: str, body=None) -> tuple[int, object]:

            auth=(client_id, client_secret),

            timeout=TIMEOUT,

        )

        assert r.status_code == 200, f"no token for {client_id} in {realm}: {r.text}"

        return r.json()["access_token"]

        assert r.status_code == 200, f"no token for {client_id} in {realm}: {r.status_code}"

        return Token(r.json()["access_token"])

    def partner_id(self, other: "Operator") -> str:

        status, partners = self.fm_internal("GET", "/internal/partners")

            },

            timeout=TIMEOUT,

        )

        assert r.status_code == 200, f"Keycloak admin login failed: {r.text}"

        return r.json()["access_token"]

        assert r.status_code == 200, f"Keycloak admin login failed: {r.status_code}"

        return Token(r.json()["access_token"])

    def _call(self, method, path, **kwargs):

        r = http(method, self.base + path, self._token(), **kwargs)

"""What an anonymous caller on the network can reach."""

import pytest

from conftest import http

@pytest.mark.xfail(

    strict=True,

    reason="FM serves /internal/* on the EWBI port without auth; the design says it must not "

    "be exposed on the EWBI path. Remove this mark once that is fixed.",

)

def test_fm_internal_api_is_not_public(operators):

    """FM serves /internal/* only on its cluster-internal port, never on the EWBI one."""

    for op in operators:

        for path in ("/internal/partners", "/internal/identity"):

            r = http("GET", op.fm_url + path)

            assert r.status_code in (401, 403, 404), f"{op.host} {path}: {r.status_code}"

            assert r.status_code == 404, f"{op.host} {path}: {r.status_code}"