feat(profile): add secrets handling to kubernetes and machine init profiles

charmcraft/templates/init-kubernetes/charmcraft.yaml.j2

@@ -48,6 +48,17 @@ config:
         Acceptable values are: "info", "debug", "warning", "error" and "critical"
       default: "info"
       type: string
+    # A secret config option for a user-provided API token.
+    # See https://documentation.ubuntu.com/juju/3.6/howto/manage-secrets/
+    api-token:
+      description: |
+        URI of a Juju user secret containing the API token for the workload.
+        The secret must have a 'token' key.
+
+        Create the secret: juju add-secret my-api-token token=<value>
+        Grant access:      juju grant-secret <id> {{ name }}
+        Set this option:   juju config {{ name }} api-token=<id>
+      type: secret
 
 # Your workload's containers.
 # https://documentation.ubuntu.com/charmcraft/stable/reference/files/charmcraft-yaml-file/#containers

charmcraft/templates/init-kubernetes/src/charm.py.j2

@@ -4,7 +4,9 @@
 
 """Charm the application."""
 
+import datetime
 import logging
+import secrets
 import time
 
 import ops
@@ -15,6 +17,8 @@ import {{ workload_module }}
 logger = logging.getLogger(__name__)
 
 SERVICE_NAME = "some-service"  # Name of Pebble service that runs in the workload container.
+API_TOKEN_LABEL = "api-token"  # Label for the user-provided secret named in charm config.
+APP_SECRET_LABEL = "workload-password"  # Label for the app-managed workload secret.
 
 
 class {{ class_name }}(ops.CharmBase):
@@ -23,6 +27,11 @@ class {{ class_name }}(ops.CharmBase):
     def __init__(self, framework: ops.Framework):
         super().__init__(framework)
         framework.observe(self.on["some_container"].pebble_ready, self._on_pebble_ready)
+        framework.observe(self.on.config_changed, self._on_config_changed)
+        framework.observe(self.on.secret_changed, self._on_secret_changed)
+        framework.observe(self.on.secret_rotate, self._on_secret_rotate)
+        framework.observe(self.on.secret_expired, self._on_secret_expired)
+        framework.observe(self.on.secret_remove, self._on_secret_remove)
         self.container = self.unit.get_container("some-container")
 
     def _on_pebble_ready(self, event: ops.PebbleReadyEvent):
@@ -49,8 +58,73 @@ class {{ class_name }}(ops.CharmBase):
         version = {{ workload_module }}.get_version()
         if version is not None:
             self.unit.set_workload_version(version)
+        # Create the app-managed workload secret on first start (leader only).
+        if self.unit.is_leader():
+            self._ensure_app_secret()
         self.unit.status = ops.ActiveStatus()
 
+    def _on_config_changed(self, event: ops.ConfigChangedEvent):
+        """Handle config-changed event — the api-token config might name a different secret."""
+        self._configure_api_token()
+
+    def _on_secret_changed(self, event: ops.SecretChangedEvent):
+        """Handle secret-changed event — the api-token secret has a new revision."""
+        if event.secret.label == API_TOKEN_LABEL:
+            self._configure_api_token()
+
+    def _configure_api_token(self) -> None:
+        """Read the user-provided api-token secret and configure the workload with it."""
+        api_token_id = self.config.get("api-token")
+        if not api_token_id:
+            return
+        try:
+            # Attaching a label lets the secret-changed handler recognise this secret.
+            secret = self.model.get_secret(id=api_token_id, label=API_TOKEN_LABEL)
+            # 'refresh' tells Juju to start tracking the latest revision.
+            content = secret.get_content(refresh=True)
+        except ops.SecretNotFoundError:
+            self.unit.status = ops.BlockedStatus("api-token secret not found or not granted")
+            return
+        token = content.get("token", "")
+        # Use 'token' to configure your workload, for example:
+        # {{ workload_module }}.configure(api_token=token)
+        logger.info("api-token resolved (%d chars); reconfiguring workload", len(token))
+        self.unit.status = ops.ActiveStatus()
+
+    def _ensure_app_secret(self) -> ops.Secret:
+        """Get or create the app-managed workload password (idempotent, leader only)."""
+        try:
+            return self.model.get_secret(label=APP_SECRET_LABEL)
+        except ops.SecretNotFoundError:
+            return self.app.add_secret(
+                {"password": secrets.token_hex(32)},
+                label=APP_SECRET_LABEL,
+                rotate=ops.SecretRotate.MONTHLY,
+                expire=datetime.timedelta(days=90),
+            )
+
+    def _on_secret_rotate(self, event: ops.SecretRotateEvent):
+        """Handle secret-rotate event — the rotation policy says it's time for a new password."""
+        if event.secret.label == APP_SECRET_LABEL:
+            self._set_new_password(event.secret)
+
+    def _on_secret_expired(self, event: ops.SecretExpiredEvent):
+        """Handle secret-expired event — the current password must not be used any more."""
+        if event.secret.label == APP_SECRET_LABEL:
+            self._set_new_password(event.secret)
+
+    def _set_new_password(self, secret: ops.Secret) -> None:
+        """Add a new revision of the app-managed workload password."""
+        # Don't remove the old revision here: other units might still be using it.
+        # Juju emits secret-remove once no unit is tracking it any more.
+        secret.set_content({"password": secrets.token_hex(32)})
+
+    def _on_secret_remove(self, event: ops.SecretRemoveEvent):
+        """Handle secret-remove event — no unit is using this revision any more."""
+        # Without this, every rotation leaves another revision behind.
+        if event.secret.label == APP_SECRET_LABEL:
+            event.remove_revision()
+
     def is_ready(self) -> bool:
         """Check whether the workload is ready to use."""
         # We'll first check whether all Pebble services are running.

charmcraft/templates/init-kubernetes/tests/integration/test_charm.py.j2

@@ -29,6 +29,39 @@ def test_deploy(charm: pathlib.Path, juju: jubilant.Juju):
     juju.wait(jubilant.all_active)
 
 
+def test_app_managed_secret_created(charm: pathlib.Path, juju: jubilant.Juju):
+    """Check that the charm created the app-managed workload password on pebble-ready."""
+    labels = [secret.label for secret in juju.secrets()]
+    assert "workload-password" in labels, (
+        "Expected the charm to create a 'workload-password' app secret on start. "
+        "Check that the leader unit ran pebble-ready successfully."
+    )
+
+
+def test_user_secret(charm: pathlib.Path, juju: jubilant.Juju):
+    """Test that the charm resolves a user-provided api-token secret."""
+    # Create a user secret with a 'token' key, grant the charm access, and
+    # point the charm at it.
+    secret_uri = juju.add_secret("test-api-token", {"token": "shortvalue"})
+    juju.grant_secret(secret_uri, "{{ name }}")
+    juju.config("{{ name }}", {"api-token": secret_uri})
+    juju.wait(jubilant.all_active)
+
+    # The charm logs the length of the token it resolved.
+    assert "api-token resolved (10 chars)" in juju.debug_log()
+
+    # Act — add a new revision of the same secret. Juju notifies the charm with
+    # a secret-changed event, and the charm re-reads the content.
+    juju.update_secret(secret_uri, {"token": "a-much-longer-token-value"})
+    juju.wait(jubilant.all_active)
+
+    # Assert — the charm picked up the new revision, not the original one.
+    assert "api-token resolved (25 chars)" in juju.debug_log()
+
+    # Clean up: reset the config option so subsequent tests start clean.
+    juju.config("{{ name }}", reset=["api-token"])
+
+
 # If you implement {{ workload_module }}.get_version in the charm source,
 # remove the @pytest.mark.skip line to enable this test.
 # Alternatively, remove this test if you don't need it.

charmcraft/templates/init-kubernetes/tests/unit/test_charm.py.j2

@@ -7,7 +7,7 @@ import ops
 import pytest
 from ops import testing
 
-from charm import SERVICE_NAME, {{ class_name }}
+from charm import API_TOKEN_LABEL, APP_SECRET_LABEL, SERVICE_NAME, {{ class_name }}
 
 CHECK_NAME = "service-ready"  # Name of Pebble check in the mock workload container.
 
@@ -43,22 +43,27 @@ def mock_get_version():
     return "1.0.0"
 
 
-def test_pebble_ready(monkeypatch: pytest.MonkeyPatch):
-    """Test that the charm has the correct state after handling the pebble-ready event."""
-    # Arrange:
-    ctx = testing.Context({{ class_name }})
+def make_container(check_status: ops.pebble.CheckStatus) -> testing.Container:
+    """Build a test container with the given check status."""
     check_in = testing.CheckInfo(
         CHECK_NAME,
         level=ops.pebble.CheckLevel.READY,
-        status=ops.pebble.CheckStatus.UP,  # Simulate the Pebble check passing.
+        status=check_status,
     )
-    container_in = testing.Container(
+    return testing.Container(
         "some-container",
         can_connect=True,
         layers={"base": MOCK_LAYER},
         service_statuses={SERVICE_NAME: ops.pebble.ServiceStatus.INACTIVE},
         check_infos={check_in},
     )
+
+
+def test_pebble_ready(monkeypatch: pytest.MonkeyPatch):
+    """Test that the charm has the correct state after handling the pebble-ready event."""
+    # Arrange:
+    ctx = testing.Context({{ class_name }})
+    container_in = make_container(ops.pebble.CheckStatus.UP)
     state_in = testing.State(containers={container_in})
     monkeypatch.setattr("charm.{{ workload_module }}.get_version", mock_get_version)
 
@@ -76,20 +81,180 @@ def test_pebble_ready_service_not_ready():
     """Test that the charm raises an error if the workload isn't ready after Pebble starts it."""
     # Arrange:
     ctx = testing.Context({{ class_name }})
-    check_in = testing.CheckInfo(
-        CHECK_NAME,
-        level=ops.pebble.CheckLevel.READY,
-        status=ops.pebble.CheckStatus.DOWN,  # Simulate the Pebble check failing.
-    )
-    container_in = testing.Container(
-        "some-container",
-        can_connect=True,
-        layers={"base": MOCK_LAYER},
-        service_statuses={SERVICE_NAME: ops.pebble.ServiceStatus.INACTIVE},
-        check_infos={check_in},
-    )
+    container_in = make_container(ops.pebble.CheckStatus.DOWN)
     state_in = testing.State(containers={container_in})
 
     # Act & assert:
     with pytest.raises(testing.errors.UncaughtCharmError):
         ctx.run(ctx.on.pebble_ready(container_in), state_in)
+
+
+def test_pebble_ready_creates_app_secret(monkeypatch: pytest.MonkeyPatch):
+    """Test that the charm creates an app-managed workload secret on pebble-ready (leader)."""
+    # Arrange:
+    ctx = testing.Context({{ class_name }})
+    container_in = make_container(ops.pebble.CheckStatus.UP)
+    state_in = testing.State(containers={container_in}, leader=True)
+    monkeypatch.setattr("charm.{{ workload_module }}.get_version", mock_get_version)
+
+    # Act:
+    state_out = ctx.run(ctx.on.pebble_ready(container_in), state_in)
+
+    # Assert — the app-owned secret should appear in the output state with a 'password' key:
+    app_secrets = [s for s in state_out.secrets if s.label == APP_SECRET_LABEL]
+    assert len(app_secrets) == 1
+    assert app_secrets[0].latest_content is not None
+    assert "password" in app_secrets[0].latest_content
+
+
+def test_pebble_ready_app_secret_already_exists(monkeypatch: pytest.MonkeyPatch):
+    """Test that pebble-ready is idempotent when the app secret already exists."""
+    # Arrange:
+    ctx = testing.Context({{ class_name }})
+    container_in = make_container(ops.pebble.CheckStatus.UP)
+    existing_secret = testing.Secret(
+        tracked_content={"password": "existing-password"},
+        label=APP_SECRET_LABEL,
+        owner="app",
+    )
+    state_in = testing.State(containers={container_in}, leader=True, secrets={existing_secret})
+    monkeypatch.setattr("charm.{{ workload_module }}.get_version", mock_get_version)
+
+    # Act:
+    state_out = ctx.run(ctx.on.pebble_ready(container_in), state_in)
+
+    # Assert — the existing secret is still present; no extra secret was created:
+    app_secrets = [s for s in state_out.secrets if s.label == APP_SECRET_LABEL]
+    assert len(app_secrets) == 1
+
+
+def test_config_changed_with_api_token():
+    """Test that config-changed resolves a user-provided secret and goes active."""
+    # Arrange:
+    ctx = testing.Context({{ class_name }})
+    secret = testing.Secret(
+        tracked_content={"token": "my-secret-api-token"},
+        owner=None,  # user-owned secret (not app- or unit-owned)
+    )
+    state_in = testing.State(
+        secrets={secret},
+        config={"api-token": secret.id},
+    )
+
+    # Act:
+    state_out = ctx.run(ctx.on.config_changed(), state_in)
+
+    # Assert:
+    assert state_out.unit_status == testing.ActiveStatus()
+
+
+def test_config_changed_api_token_not_found():
+    """Test that config-changed blocks when the secret URI is not accessible."""
+    # Arrange:
+    ctx = testing.Context({{ class_name }})
+    state_in = testing.State(
+        config={"api-token": "secret:doesnotexist00000000"},
+        # No matching secret in state — simulates the secret not being granted.
+    )
+
+    # Act:
+    state_out = ctx.run(ctx.on.config_changed(), state_in)
+
+    # Assert:
+    assert isinstance(state_out.unit_status, testing.BlockedStatus)
+
+
+def test_config_changed_no_token():
+    """Test that config-changed without api-token set does nothing."""
+    # Arrange:
+    ctx = testing.Context({{ class_name }})
+    state_in = testing.State()
+
+    # Act:
+    state_out = ctx.run(ctx.on.config_changed(), state_in)
+
+    # Assert — status unchanged (charm hasn't done anything yet):
+    assert not isinstance(state_out.unit_status, testing.BlockedStatus)
+
+
+def test_secret_rotate():
+    """Test that secret-rotate generates a new revision of the workload password."""
+    # Arrange:
+    ctx = testing.Context({{ class_name }})
+    secret = testing.Secret(
+        tracked_content={"password": "old-password"},
+        label=APP_SECRET_LABEL,
+        owner="app",
+    )
+    state_in = testing.State(secrets={secret}, leader=True)
+
+    # Act:
+    state_out = ctx.run(ctx.on.secret_rotate(secret), state_in)
+
+    # Assert — a new revision with a different password should have been added:
+    out_secret = state_out.get_secret(label=APP_SECRET_LABEL)
+    assert out_secret.latest_content is not None
+    assert "password" in out_secret.latest_content
+    assert out_secret.latest_content["password"] != "old-password"
+
+
+def test_secret_expired():
+    """Test that secret-expired replaces the expired password with a new revision."""
+    # Arrange:
+    ctx = testing.Context({{ class_name }})
+    secret = testing.Secret(
+        tracked_content={"password": "expired-password"},
+        label=APP_SECRET_LABEL,
+        owner="app",
+    )
+    state_in = testing.State(secrets={secret}, leader=True)
+
+    # Act — simulate the tracked revision expiring:
+    state_out = ctx.run(ctx.on.secret_expired(secret, revision=1), state_in)
+
+    # Assert — a new revision should have been added, not the old one removed:
+    out_secret = state_out.get_secret(label=APP_SECRET_LABEL)
+    assert out_secret.latest_content is not None
+    assert out_secret.latest_content["password"] != "expired-password"
+
+
+def test_secret_changed_new_api_token_revision():
+    """Test that the charm picks up a new revision of the user-provided api-token."""
+    # Arrange:
+    ctx = testing.Context({{ class_name }})
+    secret = testing.Secret(
+        tracked_content={"token": "old-api-token"},
+        latest_content={"token": "new-api-token"},
+        label=API_TOKEN_LABEL,
+        owner=None,  # user-owned secret (not app- or unit-owned)
+    )
+    state_in = testing.State(secrets={secret}, config={"api-token": secret.id})
+
+    # Act:
+    state_out = ctx.run(ctx.on.secret_changed(secret), state_in)
+
+    # Assert — the charm refreshed, so it now tracks the latest revision:
+    out_secret = state_out.get_secret(label=API_TOKEN_LABEL)
+    assert out_secret.tracked_content == {"token": "new-api-token"}
+    assert state_out.unit_status == testing.ActiveStatus()
+
+
+def test_secret_remove():
+    """Test that secret-remove cleans up a revision that no unit is using."""
+    # Arrange — rotate twice, so that revision 2 is neither the oldest revision
+    # nor the newest one:
+    ctx = testing.Context({{ class_name }})
+    secret = testing.Secret(
+        tracked_content={"password": "first-password"},
+        label=APP_SECRET_LABEL,
+        owner="app",
+    )
+    state = testing.State(secrets={secret}, leader=True)
+    for _ in range(2):
+        state = ctx.run(ctx.on.secret_rotate(state.get_secret(label=APP_SECRET_LABEL)), state)
+
+    # Act — Juju reports that revision 2 is no longer tracked by any unit:
+    ctx.run(ctx.on.secret_remove(state.get_secret(label=APP_SECRET_LABEL), revision=2), state)
+
+    # Assert — the charm asked Juju to remove that revision:
+    assert ctx.removed_secret_revisions == [2]