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
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
"""API tokens: the credential a machine caller presents instead of a cookie.

The interesting properties are the ones that stop a leaked token from being
worth more than the account it belongs to - it cannot be presented under another
address, it cannot outlive a revocation, and it cannot exceed its user's role.
"""
import base64

import pytest

from app.core import auth as auth_core
from app.core.context import Forbidden, NotFound, Principal, Scope
from db.db_declarations import ApiToken, Organization

PASSWORD = "correcthorsebatterystaple"


@pytest.fixture
def owner(session):
    """An owner of one organisation, and the Principal for them."""
    org = Organization(name="Acme", slug="acme")
    session.add(org)
    session.commit()
    user = auth_core.create_user(session, "dev@example.com", PASSWORD, "Dev")
    auth_core.add_membership(session, user, org, "owner")
    return user, org, auth_core.principal_for_user(session, user)


def test_the_secret_is_returned_once_and_never_stored(session, owner):
    _, _, principal = owner
    token, secret = auth_core.create_token(session, principal, "CI")

    assert secret.startswith(auth_core.TOKEN_PREFIX)
    assert secret not in token.token_hash
    assert len(token.token_hash) == 64
    # The prefix is for telling two apart in a list, and is short enough not to
    # help reconstruct either.
    assert secret.startswith(token.prefix)
    assert "secret" not in token.to_dict()


def test_a_token_authenticates(session, owner):
    user, org, principal = owner
    _, secret = auth_core.create_token(session, principal, "CI")

    resolved = auth_core.authenticate_basic(session, user.email, secret)
    assert resolved.user_id == user.id
    assert resolved.org_id == org.id
    assert resolved.surface == "api"


def test_a_token_carries_its_users_role_and_no_more(session):
    """A viewer's token reads and does not write. The token is the person's
    credential, so it cannot be a way to exceed what the person may do."""
    org = Organization(name="Acme", slug="acme")
    session.add(org)
    session.commit()
    user = auth_core.create_user(session, "viewer@example.com", PASSWORD)
    auth_core.add_membership(session, user, org, "viewer")

    # A viewer cannot mint one at all: creating a token needs WRITE.
    viewer = auth_core.principal_for_user(session, user)
    with pytest.raises(Forbidden):
        auth_core.create_token(session, viewer, "nope")
    assert Scope.READ in viewer.scopes
    assert Scope.WRITE not in viewer.scopes


def test_a_token_cannot_be_presented_under_another_address(session, owner):
    """The email half is not decorative: without this check a leaked token could
    be replayed under a different, more privileged account's name."""
    _, _, principal = owner
    _, secret = auth_core.create_token(session, principal, "CI")
    auth_core.create_user(session, "someone-else@example.com", PASSWORD)

    with pytest.raises(Forbidden):
        auth_core.authenticate_basic(session, "someone-else@example.com", secret)


def test_a_revoked_token_stops_working_but_is_kept(session, owner):
    user, _, principal = owner
    token, secret = auth_core.create_token(session, principal, "CI")

    auth_core.revoke_token(session, principal, token.id)
    with pytest.raises(Forbidden):
        auth_core.authenticate_basic(session, user.email, secret)

    # Revocation is a timestamp, not a delete.
    assert session.query(ApiToken).count() == 1
    assert session.query(ApiToken).one().revoked_at is not None


def test_an_expired_token_is_refused(session, owner):
    from datetime import datetime, timedelta

    user, _, principal = owner
    token, secret = auth_core.create_token(
        session, principal, "CI", expires_at=datetime.utcnow() - timedelta(seconds=1))
    with pytest.raises(Forbidden):
        auth_core.authenticate_basic(session, user.email, secret)


def test_a_pending_account_cannot_use_its_token(session, owner):
    """Unreachable by the front door today, because minting needs a session. It
    is checked because approval can be withdrawn after a token was minted."""
    user, _, principal = owner
    _, secret = auth_core.create_token(session, principal, "CI")

    user.approved_at = None
    session.commit()
    with pytest.raises(Forbidden):
        auth_core.authenticate_basic(session, user.email, secret)


def test_garbage_is_refused_without_a_lookup(session, owner):
    user, _, _ = owner
    for secret in ("", "not-a-token", "Bearer something"):
        with pytest.raises(Forbidden):
            auth_core.authenticate_basic(session, user.email, secret)


def test_revoking_someone_elses_token_is_not_found(session, owner):
    _, org, principal = owner
    token, _ = auth_core.create_token(session, principal, "CI")

    other = auth_core.create_user(session, "other@example.com", PASSWORD)
    auth_core.add_membership(session, other, org, "owner")
    other_principal = auth_core.principal_for_user(session, other)

    with pytest.raises(NotFound):
        auth_core.revoke_token(session, other_principal, token.id)


def test_list_is_scoped_to_the_caller(session, owner):
    _, org, principal = owner
    auth_core.create_token(session, principal, "Mine")

    other = auth_core.create_user(session, "other@example.com", PASSWORD)
    auth_core.add_membership(session, other, org, "owner")
    other_principal = auth_core.principal_for_user(session, other)
    auth_core.create_token(session, other_principal, "Theirs")

    assert [t.name for t in auth_core.list_tokens(session, principal)] == ["Mine"]
    assert [t.name for t in auth_core.list_tokens(session, other_principal)] == ["Theirs"]


# --- the façade -------------------------------------------------------------


def test_the_basic_surface_skips_the_session_guard(account):
    """Everything under the Basic prefix authenticates itself, so the guard lets
    it through without a cookie - and the endpoint, not the guard, is what
    refuses an unauthenticated caller.

    The skeleton registers no route under that prefix yet, so a 404 here is the
    right answer: it proves the guard did not redirect to /login first.
    """
    client, _, _ = account
    client.get("/logout")
    r = client.get("/api/v1/anything")
    assert r.status_code == 404


def test_minting_is_refused_on_a_plan_without_it(account):
    """The paywall, end to end: 402 and not 403, because it is fixed by the
    customer upgrading rather than by an administrator granting anything."""
    from tests.conftest import csrf_for

    client, _, _ = account
    r = client.post("/api/tokens", json={"name": "CI"},
                    headers={"X-CSRF-Token": csrf_for(client)})
    assert r.status_code == 402, r.get_json()
    assert "upgrade" in r.get_json()["error"].lower()


def test_listing_and_revoking_survive_a_lapsed_plan(account):
    """Minting is gated; seeing and revoking what you already hold is not. Losing
    the ability to turn off a credential you cannot see is how a leaked token
    outlives the subscription that made it."""
    from tests.conftest import csrf_for, on_tier

    client, _, org = account
    on_tier(org, "silver")
    csrf = csrf_for(client)
    created = client.post("/api/tokens", json={"name": "CI"},
                          headers={"X-CSRF-Token": csrf})
    token_id = created.get_json()["token"]["id"]

    on_tier(org, "free")
    assert client.get("/api/tokens").status_code == 200
    r = client.delete(f"/api/tokens/{token_id}", headers={"X-CSRF-Token": csrf})
    assert r.status_code == 200, r.get_json()


def test_a_token_round_trips_through_http(account):
    from utils.db_wrapper import db_session

    client, user, org = account
    from tests.conftest import csrf_for, on_tier
    # Minting is a Silver feature - see plans.FEATURES and the gate in
    # endpoints/session.api_token_create.
    on_tier(org, "silver")
    csrf = csrf_for(client)

    r = client.post("/api/tokens", json={"name": "CI"},
                    headers={"X-CSRF-Token": csrf})
    assert r.status_code == 200, r.get_json()
    secret = r.get_json()["secret"]

    # The secret authenticates as Basic, exactly as a script would send it.
    pair = base64.b64encode(f"{user.email}:{secret}".encode()).decode()
    resolved = auth_core.authenticate_basic(db_session, user.email, secret)
    assert resolved.user_id == user.id
    assert pair  # the header a caller would build from it

    listed = client.get("/api/tokens").get_json()["tokens"]
    assert [t["name"] for t in listed] == ["CI"]
    # Listing never returns the secret.
    assert "secret" not in listed[0]