agilentics / boiler 
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
"""Google sign-in, without talking to Google.

The identity resolution (app/core/oauth) is tested directly; the endpoint is
tested up to the point where it would make a network call. What is deliberately
covered is the security-relevant half - unset means off, state must match, an
unverified email is refused - because that is what a later edit is most likely to
break silently.
"""
import pytest

import settings
from app.core import auth as auth_core
from app.core import oauth as oauth_core
from db.db_declarations import Organization, User

PASSWORD = "correcthorsebatterystaple"


@pytest.fixture
def google_configured(monkeypatch):
    monkeypatch.setattr(settings.Config, "GOOGLE_CLIENT_ID", "client-id.apps.googleusercontent.com")
    monkeypatch.setattr(settings.Config, "GOOGLE_CLIENT_SECRET", "client-secret")


# --- configuration gate ------------------------------------------------------


def test_unset_config_means_the_feature_is_off(app_client):
    """The stance the whole app takes: no credential registered anywhere, and
    everything still runs. Both routes bounce rather than erroring."""
    from app.endpoints.oauth import is_enabled

    assert not is_enabled()
    for path in ("/auth/google/start", "/auth/google/callback"):
        r = app_client.get(path)
        assert r.status_code == 302
        assert r.headers["Location"].endswith("/login")


def test_the_button_is_hidden_when_unconfigured(app_client):
    html = app_client.get("/login").get_data(as_text=True)
    assert "/auth/google/start" not in html


def test_the_button_appears_when_configured(app_client, google_configured):
    html = app_client.get("/login").get_data(as_text=True)
    assert "/auth/google/start" in html


def test_the_oauth_paths_are_public(app_client):
    from app.endpoints.crud import is_public

    assert is_public("/auth/google/start")
    assert is_public("/auth/google/callback")


# --- the redirect dance ------------------------------------------------------


def test_start_redirects_to_google_and_stashes_state(app_client, google_configured):
    from app.endpoints.oauth import NONCE_KEY, STATE_KEY

    r = app_client.get("/auth/google/start")
    assert r.status_code == 302

    location = r.headers["Location"]
    assert location.startswith("https://accounts.google.com/o/oauth2/v2/auth")
    assert "response_type=code" in location
    assert "prompt=select_account" in location
    # Built from the request only because the test client's host is loopback;
    # any other host without PUBLIC_BASE_URL fails closed. See crud.public_base_url.
    assert "redirect_uri=http%3A%2F%2Flocalhost%2Fauth%2Fgoogle%2Fcallback" in location

    with app_client.session_transaction() as sess:
        assert sess[STATE_KEY] and sess[NONCE_KEY]
        assert sess[STATE_KEY] in location


def test_a_callback_with_no_state_is_refused(app_client, google_configured):
    """A callback nobody started - the forged-request case `state` exists for."""
    r = app_client.get("/auth/google/callback?code=abc&state=made-up")
    assert r.status_code == 302
    assert r.headers["Location"].endswith("/login?error=google")


def test_a_callback_whose_state_does_not_match_is_refused(app_client, google_configured):
    from app.endpoints.oauth import STATE_KEY

    with app_client.session_transaction() as sess:
        sess[STATE_KEY] = "the-real-state"

    r = app_client.get("/auth/google/callback?code=abc&state=not-the-real-state")
    assert r.headers["Location"].endswith("/login?error=google")


def test_state_is_single_use(app_client, google_configured):
    """Popped rather than read, so a replayed callback finds nothing to match."""
    from app.endpoints.oauth import STATE_KEY

    with app_client.session_transaction() as sess:
        sess[STATE_KEY] = "state-x"

    # No code, so it fails after the state check - which is what consumes it.
    app_client.get("/auth/google/callback?state=state-x")

    with app_client.session_transaction() as sess:
        assert STATE_KEY not in sess


def test_consent_denied_returns_to_login(app_client, google_configured):
    r = app_client.get("/auth/google/callback?error=access_denied")
    assert r.headers["Location"].endswith("/login?error=google")


# --- identity resolution -----------------------------------------------------


def test_a_new_google_user_gets_an_account_and_a_tenant(session):
    user = oauth_core.find_or_create_google_user(
        session, google_sub="sub-1", email="Ada@Example.com", name="Ada Lovelace")

    assert user.email == "ada@example.com"          # normalised
    assert user.google_sub == "sub-1"
    assert user.password_hash is None               # provider-only account

    p = auth_core.principal_for_user(session, user)
    org = session.get(Organization, p.org_id)
    assert org.slug == "ada-lovelace"
    assert session.query(Organization).count() == 1


def test_signing_in_again_returns_the_same_account(session):
    first = oauth_core.find_or_create_google_user(
        session, google_sub="sub-1", email="ada@example.com", name="Ada")
    again = oauth_core.find_or_create_google_user(
        session, google_sub="sub-1", email="ada@example.com", name="Ada")

    assert first.id == again.id
    assert session.query(User).count() == 1
    assert session.query(Organization).count() == 1


def test_google_links_to_an_existing_password_account(session):
    """Safe only because the endpoint refuses a claim without email_verified -
    otherwise this is a way to seize somebody's account."""
    org = Organization(name="Acme", slug="acme")
    session.add(org)
    session.commit()
    existing = auth_core.create_user(session, "dev@example.com", PASSWORD)
    auth_core.add_membership(session, existing, org, "member")

    linked = oauth_core.find_or_create_google_user(
        session, google_sub="sub-9", email="dev@example.com")

    assert linked.id == existing.id
    assert linked.google_sub == "sub-9"
    # The password still works: linking adds a way in, it does not replace one.
    assert auth_core.authenticate_password(session, "dev@example.com", PASSWORD)
    # And no second organisation was provisioned for an account that had one.
    assert session.query(Organization).count() == 1


def test_a_second_person_of_the_same_name_gets_a_free_slug(session):
    oauth_core.find_or_create_google_user(
        session, google_sub="sub-1", email="ada@one.com", name="Ada")
    oauth_core.find_or_create_google_user(
        session, google_sub="sub-2", email="ada@two.com", name="Ada")

    slugs = {o.slug for o in session.query(Organization).all()}
    assert slugs == {"ada", "ada-2"}


def test_a_name_that_does_not_slugify_still_provisions(session):
    """A display name of punctuation, or none at all, must not produce a slug the
    validator rejects - this path never asked anyone for one."""
    user = oauth_core.find_or_create_google_user(
        session, google_sub="sub-1", email="someone@example.com", name="!!!")
    p = auth_core.principal_for_user(session, user)
    assert session.get(Organization, p.org_id).slug == "someone"


def test_a_malformed_claim_pair_is_a_programming_error(session):
    with pytest.raises(ValueError):
        oauth_core.find_or_create_google_user(session, google_sub="", email="a@b.com")
    with pytest.raises(ValueError):
        oauth_core.find_or_create_google_user(session, google_sub="s", email="not-an-email")