Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Secret ownership fixes #86

Merged
merged 14 commits into from
Jan 9, 2024
16 changes: 10 additions & 6 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -811,11 +811,16 @@ state = State(
```

The only mandatory arguments to Secret are its secret ID (which should be unique) and its 'contents': that is, a mapping
from revision numbers (integers) to a str:str dict representing the payload of the revision.
from revision numbers (integers) to a `str:str` dict representing the payload of the revision.

By default, the secret is not owned by **this charm** nor is it granted to it.
Therefore, if charm code attempted to get that secret revision, it would get a permission error: we didn't grant it to
this charm, nor we specified that the secret is owned by it.
There are three cases:
- the secret is owned by this app, in which case only the leader unit can manage it
- the secret is owned by this unit, in which case this charm can always manage it (leader or not)
- (default) the secret is not owned by this app nor unit, which means we can't manage it but only view it

Thus by default, the secret is not owned by **this charm**, but, implicitly, by some unknown 'other charm', and that other charm has granted us view rights.

The presence of the secret in `State.secrets` entails that we have access to it, either as owners or as grantees. Therefore, if we're not owners, we must be grantees. Absence of a Secret from the known secrets list means we are not entitled to obtaining it in any way. The charm, indeed, shouldn't even know it exists.

To specify a secret owned by this unit (or app):

Expand All @@ -826,7 +831,7 @@ state = State(
secrets=[
Secret(
id='foo',
contents={0: {'key': 'public'}},
contents={0: {'key': 'private'}},
owner='unit', # or 'app'
remote_grants={0: {"remote"}}
# the secret owner has granted access to the "remote" app over some relation with ID 0
Expand All @@ -846,7 +851,6 @@ state = State(
id='foo',
contents={0: {'key': 'public'}},
# owner=None, which is the default
granted="unit", # or "app",
revision=0, # the revision that this unit (or app) is currently tracking
)
]
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "ops-scenario"

version = "5.7"
version = "5.7.1"

authors = [
{ name = "Pietro Pasotti", email = "[email protected]" }
Expand Down
37 changes: 27 additions & 10 deletions scenario/mocking.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
Event,
ExecOutput,
Relation,
Secret,
State,
SubordinateRelation,
_CharmSpec,
Expand Down Expand Up @@ -302,6 +303,19 @@ def secret_add(
self._state.secrets.append(secret)
return id

def _check_can_manage_secret(
self,
secret: "Secret",
):
# FIXME: match real tracebacks
if secret.owner is None:
raise SecretNotFoundError("this secret is not owned by this unit/app")
PietroPasotti marked this conversation as resolved.
Show resolved Hide resolved
if secret.owner == "app" and not self.is_leader():
raise SecretNotFoundError(
PietroPasotti marked this conversation as resolved.
Show resolved Hide resolved
f"App-owned secret {secret.id!r} can only be "
f"managed by the leader.",
)

def secret_get(
self,
*,
Expand All @@ -311,6 +325,12 @@ def secret_get(
peek: bool = False,
) -> Dict[str, str]:
secret = self._get_secret(id, label)

if self._context.juju_version <= "3.2":
# in juju<3.2, secret owners always track the latest revision.
PietroPasotti marked this conversation as resolved.
Show resolved Hide resolved
if secret.owner is not None:
refresh = True

revision = secret.revision
if peek or refresh:
revision = max(secret.contents.keys())
Expand All @@ -326,8 +346,9 @@ def secret_info_get(
label: Optional[str] = None,
) -> SecretInfo:
secret = self._get_secret(id, label)
if not secret.owner:
raise RuntimeError(f"not the owner of {secret}")

# only "manage"=write access level can read secret info
self._check_can_manage_secret(secret)

return SecretInfo(
id=secret.id,
Expand All @@ -349,8 +370,7 @@ def secret_set(
rotate: Optional[SecretRotate] = None,
):
secret = self._get_secret(id, label)
if not secret.owner:
raise RuntimeError(f"not the owner of {secret}")
self._check_can_manage_secret(secret)

secret._update_metadata(
content=content,
Expand All @@ -362,8 +382,7 @@ def secret_set(

def secret_grant(self, id: str, relation_id: int, *, unit: Optional[str] = None):
secret = self._get_secret(id)
if not secret.owner:
raise RuntimeError(f"not the owner of {secret}")
self._check_can_manage_secret(secret)

grantee = unit or self._get_relation_by_id(relation_id).remote_app_name

Expand All @@ -374,16 +393,14 @@ def secret_grant(self, id: str, relation_id: int, *, unit: Optional[str] = None)

def secret_revoke(self, id: str, relation_id: int, *, unit: Optional[str] = None):
secret = self._get_secret(id)
if not secret.owner:
raise RuntimeError(f"not the owner of {secret}")
self._check_can_manage_secret(secret)

grantee = unit or self._get_relation_by_id(relation_id).remote_app_name
secret.remote_grants[relation_id].remove(grantee)

def secret_remove(self, id: str, *, revision: Optional[int] = None):
secret = self._get_secret(id)
if not secret.owner:
raise RuntimeError(f"not the owner of {secret}")
self._check_can_manage_secret(secret)

if revision:
del secret.contents[revision]
Expand Down
30 changes: 24 additions & 6 deletions scenario/state.py
Original file line number Diff line number Diff line change
Expand Up @@ -135,10 +135,11 @@ class Secret(_DCBase):
contents: Dict[int, "RawSecretRevisionContents"]

# indicates if the secret is owned by THIS unit, THIS app or some other app/unit.
owner: Literal["unit", "application", None] = None
# if None, the implication is that the secret has been granted to this unit.
owner: Literal["unit", "app", None] = None

# has this secret been granted to this unit/app or neither? Only applicable if NOT owner
granted: Literal["unit", "app", False] = False
# deprecated!
granted: Literal["unit", "app", False] = "<DEPRECATED>"

# what revision is currently tracked by this charm. Only meaningful if owner=False
revision: int = 0
Expand All @@ -152,6 +153,21 @@ class Secret(_DCBase):
expire: Optional[datetime.datetime] = None
rotate: SecretRotate = SecretRotate.NEVER

def __post_init__(self):
if self.granted != "<DEPRECATED>":
logger.warning(
"``state.Secret.granted`` is deprecated and will be removed in Scenario 6. "
"If a Secret is not owned by the app/unit you are testing, nor has been granted to "
"it by the (remote) owner, then omit it from ``State.secrets`` altogether.",
)
PietroPasotti marked this conversation as resolved.
Show resolved Hide resolved
if self.owner == "application":
logger.warning(
"Secret.owner='application' is deprecated in favour of 'app' "
"and will be removed in Scenario 6.",
)
# bypass frozen dataclass
object.__setattr__(self, "owner", "app")

# consumer-only events
@property
def changed_event(self):
Expand Down Expand Up @@ -205,8 +221,9 @@ def _update_metadata(
):
"""Update the metadata."""
revision = max(self.contents.keys())
self.contents[revision + 1] = content

# bypass frozen dataclass
object.__setattr__(self, "contents"[revision + 1], content)
if label:
object.__setattr__(self, "label", label)
if description:
Expand Down Expand Up @@ -822,10 +839,11 @@ class State(_DCBase):
model: Model = Model()
"""The model this charm lives in."""
secrets: List[Secret] = dataclasses.field(default_factory=list)
"""The secrets this charm has access to (as an owner, or as a grantee)."""
"""The secrets this charm has access to (as an owner, or as a grantee).
The presence of a secret in this list entails that the charm can read it.
Whether it can manage it or not depends on the individual secret's `owner` flag."""
resources: Dict[str, "PathLike"] = dataclasses.field(default_factory=dict)
"""Mapping from resource name to path at which the resource can be found."""

planned_units: int = 1
"""Number of non-dying planned units that are expected to be running this application.
Use with caution."""
Expand Down
Loading