hazza/jupyterhub
1
0
Fork 0
mirror of https://github.com/jupyterhub/jupyterhub.git synced 2025-10-07 10:04:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #3713 from minrk/custom-scopes

Browse Source 
allow user-defined custom scopes
This commit is contained in:
Erik Sundell
2022-03-16 08:52:55 +01:00
committed by GitHub
parent fa8cd90793 9a87b59e84
commit 454e356e4d
12 changed files with 623 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

113
docs/source/rbac/scopes.md
View File

@@ -119,6 +119,119 @@ Note that only the {ref}`horizontal filtering <horizontal-filtering-target>` can
Metascopes `self` and `all`, `<resource>`, `<resource>:<subresource>`, `read:<resource>`, `admin:<resource>`, and `access:<resource>` scopes are predefined and cannot be changed otherwise.
```
### Custom scopes
:::{versionadded} 2.3
:::
JupyterHub 2.3 introduces support for custom scopes.
Services that use JupyterHub for authentication and want to implement their own granular access may define additional _custom_ scopes and assign them to users with JupyterHub roles.
% Note: keep in sync with pattern/description in jupyterhub/scopes.py
Custom scope names must start with `custom:`
and contain only lowercase ascii letters, numbers, hyphen, underscore, colon, and asterisk (`-_:*`).
The part after `custom:` must start with a letter or number.
Scopes may not end with a hyphen or colon.
The only strict requirement is that a custom scope definition must have a `description`.
It _may_ also have `subscopes` if you are defining multiple scopes that have a natural hierarchy,
For example:
```python
c.JupyterHub.custom_scopes = {
"custom:myservice:read": {
"description": "read-only access to myservice",
},
"custom:myservice:write": {
"description": "write access to myservice",
# write permission implies read permission
"subscopes": [
"custom:myservice:read",
],
},
}
c.JupyterHub.load_roles = [
# graders have read-only access to the service
{
"name": "service-user",
"groups": ["graders"],
"scopes": [
"custom:myservice:read",
"access:service!service=myservice",
],
},
# instructors have read and write access to the service
{
"name": "service-admin",
"groups": ["instructors"],
"scopes": [
"custom:myservice:write",
"access:service!service=myservice",
],
},
]
```
In the above configuration, two scopes are defined:
- `custom:myservice:read` grants read-only access to the service, and
- `custom:myservice:write` grants write access to the service
- write access _implies_ read access via the `subscope`
These custom scopes are assigned to two groups via `roles`:
- users in the group `graders` are granted read access to the service
- users in the group `instructors` are
- both are granted _access_ to the service via `access:service!service=myservice`
When the service completes OAuth, it will retrieve the user model from `/hub/api/user`.
This model includes a `scopes` field which is a list of authorized scope for the request,
which can be used.
```python
def require_scope(scope):
"""decorator to require a scope to perform an action"""
def wrapper(func):
@functools.wraps(func)
def wrapped_func(request):
user = fetch_hub_api_user(request.token)
if scope not in user["scopes"]:
raise HTTP403(f"Requires scope {scope}")
else:
return func()
return wrapper
@require_scope("custom:myservice:read")
async def read_something(request):
...
@require_scope("custom:myservice:write")
async def write_something(request):
...
```
If you use {class}`~.HubOAuthenticated`, this check is performed automatically
against the `.hub_scopes` attribute of each Handler
(the default is populated from `$JUPYTERHUB_OAUTH_SCOPES` and usually `access:services!service=myservice`).
```python
from tornado import web
from jupyterhub.services.auth import HubOAuthenticated
class MyHandler(HubOAuthenticated, BaseHandler):
hub_scopes = ["custom:myservice:read"]
@web.authenticated
def get(self):
...
```
Existing scope filters (`!user=`, etc.) may be applied to custom scopes.
Custom scope _filters_ are NOT supported.
### Scopes and APIs
The scopes are also listed in the [](../reference/rest-api.rst) documentation. Each API endpoint has a list of scopes which can be used to access the API; if no scopes are listed, the API is not authenticated and can be accessed without any permissions (i.e., no scopes).

135
examples/custom-scopes/grades.py Normal file
View File

@@ -0,0 +1,135 @@
import os
from functools import wraps
from html import escape
from urllib.parse import urlparse
from tornado.httpserver import HTTPServer
from tornado.ioloop import IOLoop
from tornado.web import Application
from tornado.web import authenticated
from tornado.web import RequestHandler
from jupyterhub.services.auth import HubOAuthCallbackHandler
from jupyterhub.services.auth import HubOAuthenticated
from jupyterhub.utils import url_path_join
SCOPE_PREFIX = "custom:grades"
READ_SCOPE = f"{SCOPE_PREFIX}:read"
WRITE_SCOPE = f"{SCOPE_PREFIX}:write"
def require_scope(scopes):
"""Decorator to require scopes
For use if multiple methods on one Handler
may want different scopes,
so class-level .hub_scopes is insufficient
(e.g. read for GET, write for POST).
"""
if isinstance(scopes, str):
scopes = [scopes]
def wrap(method):
"""The actual decorator"""
@wraps(method)
@authenticated
def wrapped(self, *args, **kwargs):
self.hub_scopes = scopes
return method(self, *args, **kwargs)
return wrapped
return wrap
class MyGradesHandler(HubOAuthenticated, RequestHandler):
# no hub_scopes, anyone with access to this service
# will be able to visit this URL
@authenticated
def get(self):
self.write("<h1>My grade</h1>")
name = self.current_user["name"]
grades = self.settings["grades"]
self.write(f"<p>My name is: {escape(name)}</p>")
if name in grades:
self.write(f"<p>My grade is: {escape(str(grades[name]))}</p>")
else:
self.write("<p>No grade entered</p>")
if READ_SCOPE in self.current_user["scopes"]:
self.write('<a href="grades/">enter grades</a>')
class GradesHandler(HubOAuthenticated, RequestHandler):
# default scope for this Handler: read-only
hub_scopes = [READ_SCOPE]
def _render(self):
grades = self.settings["grades"]
self.write("<h1>All grades</h1>")
self.write("<table>")
self.write("<tr><th>Student</th><th>Grade</th></tr>")
for student, grade in grades.items():
qstudent = escape(student)
qgrade = escape(str(grade))
self.write(
f"""
<tr>
<td class="student">{qstudent}</td>
<td class="grade">{qgrade}</td>
</tr>
"""
)
if WRITE_SCOPE in self.current_user["scopes"]:
self.write("Enter grade:")
self.write(
"""
<form action=. method=POST>
<input name=student placeholder=student></input>
<input kind=number name=grade placeholder=grade></input>
<input type="submit" value="Submit">
"""
)
@require_scope([READ_SCOPE])
async def get(self):
self._render()
# POST requires WRITE_SCOPE instead of READ_SCOPE
@require_scope([WRITE_SCOPE])
async def post(self):
name = self.get_argument("student")
grade = self.get_argument("grade")
self.settings["grades"][name] = grade
self._render()
def main():
base_url = os.environ['JUPYTERHUB_SERVICE_PREFIX']
app = Application(
[
(base_url, MyGradesHandler),
(url_path_join(base_url, 'grades/'), GradesHandler),
(
url_path_join(base_url, 'oauth_callback'),
HubOAuthCallbackHandler,
),
],
cookie_secret=os.urandom(32),
grades={"student": 53},
)
http_server = HTTPServer(app)
url = urlparse(os.environ['JUPYTERHUB_SERVICE_URL'])
http_server.listen(url.port, url.hostname)
try:
IOLoop.current().start()
except KeyboardInterrupt:
pass
if __name__ == '__main__':
main()

49
examples/custom-scopes/jupyterhub_config.py Normal file
View File

@@ -0,0 +1,49 @@
import sys
c = get_config() # noqa
c.JupyterHub.services = [
{
'name': 'grades',
'url': 'http://127.0.0.1:10101',
'command': [sys.executable, './grades.py'],
'oauth_roles': ['grader'],
},
]
c.JupyterHub.custom_scopes = {
"custom:grades:read": {
"description": "read-access to all grades",
},
"custom:grades:write": {
"description": "Enter new grades",
"subscopes": ["custom:grades:read"],
},
}
c.JupyterHub.load_roles = [
{
"name": "user",
# grant all users access to services
"scopes": ["access:services", "self"],
},
{
"name": "grader",
# grant graders access to write grades
"scopes": ["custom:grades:write"],
"users": ["grader"],
},
{
"name": "instructor",
# grant instructors access to read, but not write grades
"scopes": ["custom:grades:read"],
"users": ["instructor"],
},
]
c.JupyterHub.allowed_users = {"instructor", "grader", "student"}
# dummy spawner and authenticator for testing, don't actually use these!
c.JupyterHub.authenticator_class = 'dummy'
c.JupyterHub.spawner_class = 'simple'
c.JupyterHub.ip = '127.0.0.1' # let's just run on localhost while dummy auth is enabled
c.JupyterHub.log_level = 10

19
jupyterhub/apihandlers/auth.py
View File

@@ -283,6 +283,21 @@ class OAuthAuthorizeHandler(OAuthHandler, BaseHandler):
raise web.HTTPError(
403, f"You do not have permission to access {client.description}"
)
# subset role names to those held by authenticating user
requested_role_names = set(role_names)
user = self.current_user
user_role_names = {role.name for role in user.roles}
allowed_role_names = requested_role_names.intersection(user_role_names)
excluded_role_names = requested_role_names.difference(allowed_role_names)
if excluded_role_names:
self.log.info(
f"Service {client.description} requested roles {','.join(role_names)}"
f" for user {self.current_user.name},"
f" granting only {','.join(allowed_role_names) or '[]'}."
)
role_names = list(allowed_role_names)
if not self.needs_oauth_confirm(self.current_user, client, role_names):
self.log.debug(
"Skipping oauth confirmation for %s accessing %s",
@@ -381,6 +396,10 @@ class OAuthAuthorizeHandler(OAuthHandler, BaseHandler):
# The scopes the user actually authorized, i.e. checkboxes
# that were selected.
scopes = self.get_arguments('scopes')
if scopes == []:
# avoid triggering default scopes (provider selects default scopes when scopes is falsy)
# when an explicit empty list is authorized
scopes = ["identify"]
# credentials we need in the validator
credentials = self.add_credentials()

36
jupyterhub/app.py
View File

@@ -17,10 +17,7 @@ from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
from datetime import timedelta
from datetime import timezone
from functools import partial
from getpass import getuser
from glob import glob
from itertools import chain
from operator import itemgetter
from textwrap import dedent
from urllib.parse import unquote
@@ -54,7 +51,6 @@ from traitlets import (
Unicode,
Integer,
Dict,
TraitError,
List,
Bool,
Any,
@@ -81,8 +77,10 @@ from .handlers.static import CacheControlStaticFilesHandler, LogoHandler
from .services.service import Service
from . import crypto
from . import dbutil, orm
from . import dbutil
from . import orm
from . import roles
from . import scopes
from .user import UserDict
from .oauth.provider import make_provider
from ._data import DATA_FILES_PATH
@@ -353,6 +351,29 @@ class JupyterHub(Application):
""",
).tag(config=True)
custom_scopes = Dict(
key_trait=Unicode(),
value_trait=Dict(
key_trait=Unicode(),
),
help="""Custom scopes to define.
For use when defining custom roles,
to grant users granular permissions
All custom scopes must have a description,
and must start with the prefix `custom:`.
For example::
custom_scopes = {
"custom:jupyter_server:read": {
"description": "read-only access to a single-user server",
},
}
""",
).tag(config=True)
config_file = Unicode('jupyterhub_config.py', help="The config file to load").tag(
config=True
)
@@ -2018,7 +2039,10 @@ class JupyterHub(Application):
db.commit()
async def init_role_creation(self):
"""Load default and predefined roles into the database"""
"""Load default and user-defined roles and scopes into the database"""
if self.custom_scopes:
self.log.info(f"Defining {len(self.custom_scopes)} custom scopes.")
scopes.define_custom_scopes(self.custom_scopes)
self.log.debug('Loading roles into database')
default_roles = roles.get_default_roles()
config_role_names = [r['name'] for r in self.load_roles]

119
jupyterhub/scopes.py
View File

@@ -11,9 +11,11 @@ identify scopes: set of expanded scopes needed for identify (whoami) endpoints
"""
import functools
import inspect
import re
import warnings
from enum import Enum
from functools import lru_cache
from textwrap import indent
import sqlalchemy as sa
from tornado import web
@@ -629,3 +631,120 @@ def describe_raw_scopes(raw_scopes, username=None):
}
)
return descriptions
# regex for custom scope
# for-humans description below
# note: scope description duplicated in docs/source/rbac/scopes.md
# update docs when making changes here
_custom_scope_pattern = re.compile(r"^custom:[a-z0-9][a-z0-9_\-\*:]+[a-z0-9_\*]$")
# custom scope pattern description
# used in docstring below and error message when scopes don't match _custom_scope_pattern
_custom_scope_description = """
Custom scopes must start with `custom:`
and contain only lowercase ascii letters, numbers, hyphen, underscore, colon, and asterisk (-_:*).
The part after `custom:` must start with a letter or number.
Scopes may not end with a hyphen or colon.
"""
def define_custom_scopes(scopes):
"""Define custom scopes
Adds custom scopes to the scope_definitions dict.
Scopes must start with `custom:`.
It is recommended to name custom scopes with a pattern like::
custom:$your-project:$action:$resource
e.g.::
custom:jupyter_server:read:contents
That makes them easy to parse and avoids collisions across projects.
`scopes` must have at least one scope definition,
and each scope definition must have a `description`,
which will be displayed on the oauth authorization page,
and _may_ have a `subscopes` list of other scopes if having one scope
should imply having other, more specific scopes.
Args:
scopes: dict
A dictionary of scope definitions.
The keys are the scopes,
while the values are dictionaries with at least a `description` field,
and optional `subscopes` field.
%s
Examples::
define_custom_scopes(
{
"custom:jupyter_server:read:contents": {
"description": "read-only access to files in a Jupyter server",
},
"custom:jupyter_server:read": {
"description": "read-only access to a Jupyter server",
"subscopes": [
"custom:jupyter_server:read:contents",
"custom:jupyter_server:read:kernels",
"...",
},
}
)
""" % indent(
_custom_scope_description, " " * 8
)
for scope, scope_definition in scopes.items():
if scope in scope_definitions and scope_definitions[scope] != scope_definition:
raise ValueError(
f"Cannot redefine scope {scope}={scope_definition}. Already have {scope}={scope_definitions[scope]}"
)
if not _custom_scope_pattern.match(scope):
# note: keep this description in sync with docstring above
raise ValueError(
f"Invalid scope name: {scope!r}.\n{_custom_scope_description}"
" and contain only lowercase ascii letters, numbers, hyphen, underscore, colon, and asterisk."
" The part after `custom:` must start with a letter or number."
" Scopes may not end with a hyphen or colon."
)
if "description" not in scope_definition:
raise ValueError(
f"scope {scope}={scope_definition} missing key 'description'"
)
if "subscopes" in scope_definition:
subscopes = scope_definition["subscopes"]
if not isinstance(subscopes, list) or not all(
isinstance(s, str) for s in subscopes
):
raise ValueError(
f"subscopes must be a list of scope strings, got {subscopes!r}"
)
for subscope in subscopes:
if subscope not in scopes:
if subscope in scope_definitions:
raise ValueError(
f"non-custom subscope {subscope} in {scope}={scope_definition} is not allowed."
f" Custom scopes may only have custom subscopes."
f" Roles should be used to assign multiple scopes together."
)
raise ValueError(
f"subscope {subscope} in {scope}={scope_definition} not found. All scopes must be defined."
)
extra_keys = set(scope_definition.keys()).difference(
["description", "subscopes"]
)
if extra_keys:
warnings.warn(
f"Ignoring unrecognized key(s) {', '.join(extra_keys)!r} in {scope}={scope_definition}",
UserWarning,
stacklevel=2,
)
app_log.info(f"Defining custom scope {scope}")
# deferred evaluation for debug-logging
app_log.debug("Defining custom scope %s=%s", scope, scope_definition)
scope_definitions[scope] = scope_definition

10
jupyterhub/tests/conftest.py
View File

@@ -26,6 +26,7 @@ Fixtures to add functionality or spawning behavior
# Copyright (c) Jupyter Development Team.
# Distributed under the terms of the Modified BSD License.
import asyncio
import copy
import inspect
import os
import sys
@@ -44,6 +45,7 @@ import jupyterhub.services.service
from . import mocking
from .. import crypto
from .. import orm
from .. import scopes
from ..roles import create_role
from ..roles import get_default_roles
from ..roles import mock_roles
@@ -456,3 +458,11 @@ def create_service_with_scopes(app, create_temp_role):
for service in temp_service:
app.db.delete(service)
app.db.commit()
@fixture
def preserve_scopes():
"""Revert any custom scopes after test"""
scope_definitions = copy.deepcopy(scopes.scope_definitions)
yield scope_definitions
scopes.scope_definitions = scope_definitions

8
jupyterhub/tests/mockservice.py
View File

@@ -23,6 +23,7 @@ from tornado import httpserver
from tornado import ioloop
from tornado import log
from tornado import web
from tornado.httputil import url_concat
from jupyterhub.services.auth import HubAuthenticated
from jupyterhub.services.auth import HubOAuthCallbackHandler
@@ -76,6 +77,13 @@ class OWhoAmIHandler(HubOAuthenticated, web.RequestHandler):
Uses OAuth login flow
"""
def get_login_url(self):
login_url = super().get_login_url()
scopes = self.get_argument("request-scope", None)
if scopes is not None:
login_url = url_concat(login_url, {"scope": scopes})
return login_url
@web.authenticated
def get(self):
self.write(self.get_current_user())

1
jupyterhub/tests/test_pages.py
View File

@@ -1134,6 +1134,7 @@ async def test_oauth_page_scope_appearance(
)
service = mockservice_url
user = create_user_with_scopes("access:services")
roles.grant_role(app.db, user, service_role)
oauth_client = (
app.db.query(orm.OAuthClient)
.filter_by(identifier=service.oauth_client_id)

9
jupyterhub/tests/test_roles.py
View File

@@ -498,7 +498,7 @@ async def test_load_roles_users(tmpdir, request, explicit_allowed_users):
@mark.role
async def test_load_roles_services(tmpdir, request):
async def test_load_roles_services(tmpdir, request, preserve_scopes):
services = [
{'name': 'idle-culler', 'api_token': 'some-token'},
{'name': 'user_service', 'api_token': 'some-other-token'},
@@ -509,6 +509,11 @@ async def test_load_roles_services(tmpdir, request):
'some-other-token': 'user_service',
'secret-token': 'admin_service',
}
custom_scopes = {
"custom:empty-scope": {
"description": "empty custom scope",
}
}
roles_to_load = [
{
'name': 'idle-culler',
@@ -518,11 +523,13 @@ async def test_load_roles_services(tmpdir, request):
'read:users:activity',
'read:servers',
'servers',
'custom:empty-scope',
],
'services': ['idle-culler'],
},
]
kwargs = {
'custom_scopes': custom_scopes,
'load_roles': roles_to_load,
'services': services,
'service_tokens': service_tokens,

80
jupyterhub/tests/test_scopes.py
View File

@@ -9,6 +9,7 @@ from tornado.httputil import HTTPServerRequest
from .. import orm
from .. import roles
from .. import scopes
from ..handlers import BaseHandler
from ..scopes import _check_scope_access
from ..scopes import _intersect_expanded_scopes
@@ -1047,3 +1048,82 @@ async def test_list_groups_filter(
for name in sorted(expected)
]
assert sorted(r.json(), key=itemgetter('name')) == expected_models
@pytest.mark.parametrize(
"custom_scopes",
[
{"custom:okay": {"description": "simple custom scope"}},
{
"custom:parent": {
"description": "parent",
"subscopes": ["custom:child"],
},
"custom:child": {"description": "child"},
},
{
"custom:extra": {
"description": "I have extra info",
"extra": "warn about me",
}
},
],
)
def test_custom_scopes(preserve_scopes, custom_scopes):
scopes.define_custom_scopes(custom_scopes)
for name, scope_def in custom_scopes.items():
assert name in scopes.scope_definitions
assert scopes.scope_definitions[name] == scope_def
# make sure describe works after registering custom scopes
scopes.describe_raw_scopes(list(custom_scopes.keys()))
@pytest.mark.parametrize(
"custom_scopes",
[
{
"read:users": {
"description": "Can't override",
},
},
{
"custom:empty": {},
},
{
"notcustom:prefix": {"descroption": "bad prefix"},
},
{
"custom:!illegal": {"descroption": "bad character"},
},
{
"custom:badsubscope": {
"description": "non-custom subscope not allowed",
"subscopes": [
"read:users",
],
},
},
{
"custom:nosubscope": {
"description": "subscope not defined",
"subscopes": [
"custom:undefined",
],
},
},
{
"custom:badsubscope": {
"description": "subscope not a list",
"subscopes": "custom:notalist",
},
"custom:notalist": {
"description": "the subscope",
},
},
],
)
def test_custom_scopes_bad(preserve_scopes, custom_scopes):
with pytest.raises(ValueError):
scopes.define_custom_scopes(custom_scopes)
assert scopes.scope_definitions == preserve_scopes

64
jupyterhub/tests/test_services_auth.py
View File

@@ -5,18 +5,20 @@ import sys
from binascii import hexlify
from unittest import mock
from urllib.parse import parse_qs
from urllib.parse import quote
from urllib.parse import urlparse
import pytest
from bs4 import BeautifulSoup
from pytest import raises
from tornado.httputil import url_concat
from .. import orm
from .. import roles
from .. import scopes
from ..services.auth import _ExpiringDict
from ..utils import url_path_join
from .mocking import public_url
from .test_api import add_user
from .utils import async_requests
from .utils import AsyncSession
@@ -226,6 +228,10 @@ async def test_hubauth_service_token(request, app, mockservice_url, scopes, allo
# requesting subset
(["admin", "user"], ["user"], ["user"]),
(["user", "token", "server"], ["token", "user"], ["token", "user"]),
(["admin", "user", "read-only"], ["read-only"], ["read-only"]),
# requesting valid subset, some not held by user
(["admin", "user"], ["admin", "user"], ["user"]),
(["admin", "user"], ["admin"], []),
],
)
async def test_oauth_service_roles(
@@ -235,6 +241,7 @@ async def test_oauth_service_roles(
client_allowed_roles,
request_roles,
expected_roles,
preserve_scopes,
):
service = mockservice_url
oauth_client = (
@@ -242,30 +249,40 @@ async def test_oauth_service_roles(
.filter_by(identifier=service.oauth_client_id)
.one()
)
scopes.define_custom_scopes(
{
"custom:jupyter_server:read:*": {
"description": "read-only access to jupyter server",
},
},
)
roles.create_role(
app.db,
{
"name": "read-only",
"description": "read-only access to servers",
"scopes": [
"access:servers",
"custom:jupyter_server:read:*",
],
},
)
oauth_client.allowed_roles = [
orm.Role.find(app.db, role_name) for role_name in client_allowed_roles
]
app.db.commit()
url = url_path_join(public_url(app, mockservice_url) + 'owhoami/?arg=x')
if request_roles:
url = url_concat(url, {"request-scope": " ".join(request_roles)})
# first request is only going to login and get us to the oauth form page
s = AsyncSession()
user = create_user_with_scopes("access:services")
roles.grant_role(app.db, user, "user")
roles.grant_role(app.db, user, "read-only")
name = user.name
s.cookies = await app.login_user(name)
r = await s.get(url)
r.raise_for_status()
# we should be looking at the oauth confirmation page
assert urlparse(r.url).path == app.base_url + 'hub/api/oauth2/authorize'
# verify oauth state cookie was set at some point
assert set(r.history[0].cookies.keys()) == {'service-%s-oauth-state' % service.name}
# submit the oauth form to complete authorization
data = {}
if request_roles:
data["scopes"] = request_roles
r = await s.post(r.url, data=data, headers={'Referer': r.url})
if expected_roles is None:
# expected failed auth, stop here
# verify expected 'invalid scope' error, not server error
@@ -275,6 +292,21 @@ async def test_oauth_service_roles(
assert r.status_code == 400
return
r.raise_for_status()
# we should be looking at the oauth confirmation page
assert urlparse(r.url).path == app.base_url + 'hub/api/oauth2/authorize'
# verify oauth state cookie was set at some point
assert set(r.history[0].cookies.keys()) == {'service-%s-oauth-state' % service.name}
page = BeautifulSoup(r.text, "html.parser")
scope_inputs = page.find_all("input", {"name": "scopes"})
scope_values = [input["value"] for input in scope_inputs]
print("Submitting request with scope values", scope_values)
# submit the oauth form to complete authorization
data = {}
if scope_values:
data["scopes"] = scope_values
r = await s.post(r.url, data=data, headers={'Referer': r.url})
r.raise_for_status()
assert r.url == url
# verify oauth cookie is set
assert 'service-%s' % service.name in set(s.cookies.keys())
@@ -374,7 +406,11 @@ async def test_oauth_access_scopes(
@pytest.mark.parametrize(
"token_roles, hits_page",
[([], True), (['writer'], True), (['writer', 'reader'], False)],
[
([], True),
(['writer'], True),
(['writer', 'reader'], False),
],
)
async def test_oauth_page_hit(
app,
@@ -390,6 +426,8 @@ async def test_oauth_page_hit(
}
service = mockservice_url
user = create_user_with_scopes("access:services", "self")
for role in test_roles.values():
roles.grant_role(app.db, user, role)
user.new_api_token()
token = user.api_tokens[0]
token.roles = [test_roles[t] for t in token_roles]