Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 26 additions & 1 deletion tests/common/db/oidc.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,12 @@

import factory

from warehouse.oidc.models import GitHubPublisher, PendingGitHubPublisher
from warehouse.oidc.models import (
GitHubPublisher,
GooglePublisher,
PendingGitHubPublisher,
PendingGooglePublisher,
)

from .accounts import UserFactory
from .base import WarehouseFactory
Expand Down Expand Up @@ -42,3 +47,23 @@ class Meta:
workflow_filename = "example.yml"
environment = "production"
added_by = factory.SubFactory(UserFactory)


class GooglePublisherFactory(WarehouseFactory):
class Meta:
model = GooglePublisher

id = factory.Faker("uuid4", cast_to=None)
email = factory.Faker("safe_email")
sub = factory.Faker("pystr", max_chars=12)


class PendingGooglePublisherFactory(WarehouseFactory):
class Meta:
model = PendingGooglePublisher

id = factory.Faker("uuid4", cast_to=None)
project_name = "fake-nonexistent-project"
email = factory.Faker("safe_email")
sub = factory.Faker("pystr", max_chars=12)
added_by = factory.SubFactory(UserFactory)
17 changes: 17 additions & 0 deletions tests/unit/oidc/models/test_core.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,23 @@ def test_check_claim_binary():
assert wrapped("foo", "foo", pretend.stub()) is True


def test_check_claim_invariant():
wrapped = _core._check_claim_invariant(True)
assert wrapped(True, True, pretend.stub()) is True
assert wrapped(False, True, pretend.stub()) is False

wrapped = _core._check_claim_invariant(False)
assert wrapped(False, False, pretend.stub()) is True
assert wrapped(True, False, pretend.stub()) is False

identity = object()
wrapped = _core._check_claim_invariant(identity)
assert wrapped(object(), object(), pretend.stub()) is False
assert wrapped(identity, object(), pretend.stub()) is False
assert wrapped(object(), identity, pretend.stub()) is False
assert wrapped(identity, identity, pretend.stub()) is True


class TestOIDCPublisher:
def test_oidc_publisher_not_default_verifiable(self):
publisher = _core.OIDCPublisher(projects=[])
Expand Down
186 changes: 186 additions & 0 deletions tests/unit/oidc/models/test_google.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,186 @@
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


import pretend
import pytest

from tests.common.db.oidc import GooglePublisherFactory, PendingGooglePublisherFactory
from warehouse.oidc.models import _core, google


class TestGooglePublisher:
def test_stringifies_as_email(self):
publisher = google.GooglePublisher(email="[email protected]")

assert str(publisher) == publisher.email

def test_google_publisher_all_known_claims(self):
assert google.GooglePublisher.all_known_claims() == {
# verifiable claims
"email",
"email_verified",
# optional verifiable claims
"sub",
# preverified claims
"iss",
"iat",
"nbf",
"exp",
"aud",
# unchecked claims
"azp",
"google",
}

def test_google_publisher_unaccounted_claims(self, monkeypatch):
publisher = google.GooglePublisher(
sub="fakesubject",
email="[email protected]",
)

scope = pretend.stub()
sentry_sdk = pretend.stub(
capture_message=pretend.call_recorder(lambda s: None),
push_scope=pretend.call_recorder(
lambda: pretend.stub(
__enter__=lambda *a: scope, __exit__=lambda *a: None
)
),
)
monkeypatch.setattr(_core, "sentry_sdk", sentry_sdk)

# We don't care if these actually verify, only that they're present.
signed_claims = {
claim_name: "fake"
for claim_name in google.GooglePublisher.all_known_claims()
}
signed_claims["fake-claim"] = "fake"
signed_claims["another-fake-claim"] = "also-fake"
assert not publisher.verify_claims(signed_claims=signed_claims)
assert sentry_sdk.capture_message.calls == [
pretend.call(
"JWT for GooglePublisher has unaccounted claims: "
"['another-fake-claim', 'fake-claim']"
)
]
assert scope.fingerprint == ["another-fake-claim", "fake-claim"]

def test_google_publisher_missing_claims(self, monkeypatch):
publisher = google.GooglePublisher(
sub="fakesubject",
email="[email protected]",
)

scope = pretend.stub()
sentry_sdk = pretend.stub(
capture_message=pretend.call_recorder(lambda s: None),
push_scope=pretend.call_recorder(
lambda: pretend.stub(
__enter__=lambda *a: scope, __exit__=lambda *a: None
)
),
)
monkeypatch.setattr(_core, "sentry_sdk", sentry_sdk)

signed_claims = {
claim_name: "fake"
for claim_name in google.GooglePublisher.all_known_claims()
}
# Pop the first signed claim, so that it's the first one to fail.
signed_claims.pop("email")
assert "email" not in signed_claims
assert publisher.__required_verifiable_claims__
assert not publisher.verify_claims(signed_claims=signed_claims)
assert sentry_sdk.capture_message.calls == [
pretend.call("JWT for GooglePublisher is missing claim: email")
]
assert scope.fingerprint == ["email"]

@pytest.mark.parametrize(
("email_verified", "valid"),
[(False, False), ("truthy-but-not-bool", False), ("", False), (True, True)],
)
def test_google_publisher_email_verified(self, email_verified, valid):
publisher = google.GooglePublisher(
sub="fakesubject",
email="[email protected]",
)

signed_claims = {
"sub": "fakesubject",
"email": "[email protected]",
"email_verified": email_verified,
}
assert publisher.verify_claims(signed_claims=signed_claims) is valid

@pytest.mark.parametrize(
("expected_sub", "actual_sub", "valid"),
[
# Both present: must match.
("fakesubject", "fakesubject", True),
("fakesubject", "wrongsubject", False),
# Publisher configured without subject: any subject is acceptable.
(None, "anysubject", True),
# Publisher configured with subject, none provided: must fail.
("fakesubject", None, False),
],
)
def test_google_publisher_sub_is_optional(self, expected_sub, actual_sub, valid):
publisher = google.GooglePublisher(
sub=expected_sub,
email="[email protected]",
)

signed_claims = {
"sub": actual_sub,
"email": "[email protected]",
"email_verified": True,
}
assert publisher.verify_claims(signed_claims=signed_claims) is valid


class TestPendingGooglePublisher:
@pytest.mark.parametrize("sub", ["fakesubject", None])
def test_reify_does_not_exist_yet(self, db_request, sub):
pending_publisher = PendingGooglePublisherFactory.create(sub=sub)
assert (
db_request.db.query(google.GooglePublisher)
.filter_by(
email=pending_publisher.email,
sub=pending_publisher.sub,
)
.one_or_none()
is None
)
publisher = pending_publisher.reify(db_request.db)

# If an OIDC publisher for this pending publisher does not already exist,
# a new one is created and the pending publisher is marked for deletion.
assert isinstance(publisher, google.GooglePublisher)
assert pending_publisher in db_request.db.deleted
assert publisher.email == pending_publisher.email
assert publisher.sub == pending_publisher.sub

@pytest.mark.parametrize("sub", ["fakesubject", None])
def test_reify_already_exists(self, db_request, sub):
existing_publisher = GooglePublisherFactory.create(sub=sub)
pending_publisher = PendingGooglePublisherFactory.create(
email=existing_publisher.email,
sub=existing_publisher.sub,
)
publisher = pending_publisher.reify(db_request.db)

# If an OIDC publisher for this pending publisher already exists,
# it is returned and the pending publisher is marked for deletion.
assert existing_publisher == publisher
assert pending_publisher in db_request.db.deleted
Original file line number Diff line number Diff line change
@@ -0,0 +1,58 @@
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Google OIDC models

Revision ID: fd0479fed881
Revises: d1771b942eb6
Create Date: 2023-05-02 17:45:43.772359
"""

import sqlalchemy as sa

from alembic import op
from sqlalchemy.dialects import postgresql

revision = "fd0479fed881"
down_revision = "d1771b942eb6"


def upgrade():
op.create_table(
"google_oidc_publishers",
sa.Column("email", sa.String(), nullable=False),
sa.Column("sub", sa.String(), nullable=True),
sa.Column("id", postgresql.UUID(as_uuid=True), nullable=False),
sa.ForeignKeyConstraint(
["id"],
["oidc_publishers.id"],
),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint("email", "sub", name="_google_oidc_publisher_uc"),
)
op.create_table(
"pending_google_oidc_publishers",
sa.Column("email", sa.String(), nullable=False),
sa.Column("sub", sa.String(), nullable=True),
sa.Column("id", postgresql.UUID(as_uuid=True), nullable=False),
sa.ForeignKeyConstraint(
["id"],
["pending_oidc_publishers.id"],
),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint("email", "sub", name="_pending_google_oidc_publisher_uc"),
)


def downgrade():
op.drop_table("pending_google_oidc_publishers")
op.drop_table("google_oidc_publishers")
3 changes: 3 additions & 0 deletions warehouse/oidc/models/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,10 +12,13 @@

from warehouse.oidc.models._core import OIDCPublisher, PendingOIDCPublisher
from warehouse.oidc.models.github import GitHubPublisher, PendingGitHubPublisher
from warehouse.oidc.models.google import GooglePublisher, PendingGooglePublisher

__all__ = [
"OIDCPublisher",
"PendingOIDCPublisher",
"PendingGitHubPublisher",
"PendingGooglePublisher",
"GitHubPublisher",
"GooglePublisher",
]
16 changes: 15 additions & 1 deletion warehouse/oidc/models/_core.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ def _check_claim_binary(binary_func):
ignoring the third.

This is used solely to make claim verification compatible with "trivial"
checks like `str.__eq__`.
comparison checks like `str.__eq__`.
"""

def wrapper(ground_truth, signed_claim, all_signed_claims):
Expand All @@ -39,6 +39,20 @@ def wrapper(ground_truth, signed_claim, all_signed_claims):
return wrapper


def _check_claim_invariant(value: Any):
"""
Wraps a fixed value comparison into a three-argument function.

This is used solely to make claim verification compatible with "invariant"
comparison checks, like "claim x is always the literal `true` value".
"""

def wrapper(ground_truth, signed_claim, all_signed_claims):
return ground_truth == signed_claim == value

return wrapper


class OIDCPublisherProjectAssociation(db.Model):
__tablename__ = "oidc_publisher_project_association"

Expand Down
Loading