Commit 686c7d98 authored by adietmue's avatar adietmue
Browse files

Improve Comments and README to guide new developers.

parent 0e2551fe
# new_pvk_tool
A new AMIV PVK tool using Eve and authenticating users with AMIVAPI.
## Running the tests
pip install pytest
## Backend
# Add the test user to the test database
mongo pvk_test --eval 'db.createUser({user:"pvkuser",pwd:"pvkpass",roles:["readWrite"]});'
The backend is implemented using [Eve](, a python
framework built on [Flask]( that allows to create REST
APIs with incredible ease (We use the same framework for *AMIVAPI*).
\ No newline at end of file
### How to help developing
To get started, you should first check out the
[Eve Quickstart]( to get an idea about
how the framework works. This should only take you around 10 minutes.
Afterwards, start with the file `` to get into the PVK Tool code and
keep the *Eve* and *Flask* docs at hand, if you need to look anything up.
If you add any features, don't forget to write tests!
But don't worry, it's very easy to set them up -- take a look at the
`tests` directory.
### Running the tests locally
- First of all, you need [MongoDB]( installed and
running locally.
Next, create the test user: Add the user `pvk_user` with password `pvk_pass`
and `readWrite` permission to the `pvk_test` database.
On Linux or similar, you can use this one-liner:
mongo pvk_test --eval 'db.createUser({user:"pvk_user",pwd:"pvk_pass",roles:["readWrite"]});'
- Secondly, set up your virtual environment, install requirements as well as
python -m venv env
source env/bin/activate
pip install -r requirements.txt
pip install pytest
- Finally, you can run the tests from your virtual environment with:
"""The main app.
"""PVK Tool Backend.
Here, the main app object is created.
The `create_app` function exists so that we can create apps with different
settings for tests (i.e. using a test database).
Next, you should check out the following files:
- ``:
The data model and `Eve` configuration.
- ``:
Authentication and Data Validation functions that are used in the model are
defined here. In particular, the interaction with AMIVAPI is handled here.
Check for the resource schema and API configuration.
from os import getcwd
......@@ -11,13 +23,21 @@ from security import APIAuth, APIValidator, only_own_signups
def create_app(settings=None):
"""Super simply bootstrapping for easier testing."""
"""Super simply bootstrapping for easier testing.
Initial settings are loaded from (the Flask `Config` object
makes this easy) and updated settings from the function call, if provided.
config = Config(getcwd())
if settings is not None:
# Create the app object
application = Eve(auth=APIAuth, validator=APIValidator, settings=config)
# Eve provides hooks at several points of the request,
# we use this do add dynamic filtering
for method in ['GET', 'PATCH', 'DELETE']:
event = getattr(application, 'on_pre_%s_signups' % method)
event += only_own_signups
"""AMIVAPI Authentication.
"""AMIVAPI Authentication, Validation and Filtering.
We use the `g` globals as intermediate storage:
- g['user'] is the user_id if the token was valid, otherwise None
- g['admin'] is True if the user is an admin, False otherwise
All functions in here implement additional restrictions for security,
e.g. that users can only see/modify their own signups.
The important bits are:
- Several functions to access AMIVAPI
We use the Flask `g` request globals to store results. This way, we don't
need to worry about sending a request twice. Furthermore, we can directly set
the values in `g` to avoid API requests during unittests.
- Authentication class working with tokens.
Take a look at the [Eve docs]( for
more info.
- Additional Validation rules
More info [here](
(This allows easier testing since we can just modify g)
from functools import wraps
......@@ -15,12 +30,28 @@ from import Validator
from flask import request, g, current_app
# Requests to AMIVAPI
def api_get(resource, where, projection):
"""Format and send a GET request to AMIVAPI. Return json data or None."""
url = requests.compat.urljoin(current_app.config['AMIVAPI_URL'], resource)
headers = {'Authorization': "Token %s" % g.token}
params = {
'where': json.dumps(where),
'projection': json.dumps(projection)
response = requests(url, params=params, headers=headers)
if response.status_code == 200:
return response.json()
def request_cache(key):
"""User as decorator: safe the function return in g[key]."""
"""Use as decorator: Cache the function return in g.key."""
def decorator(f):
def wrapper(*args, **kwargs):
# If the value is already in g, don't call function
return getattr(g, key)
except KeyError:
setattr(g, key, f(*args, **kwargs))
......@@ -29,23 +60,10 @@ def request_cache(key):
return decorator
def api_get(resource, token, where, projection):
"""Format and send a GET request to AMIVAPI. Return json data or None."""
url = requests.compat.urljoin(current_app.config['AMIVAPI_URL'], resource)
headers = {'Authorization': "Token %s" % token}
params = {
'where': json.dumps(where),
'projection': json.dumps(projection)
response = requests(url, params=params, headers=auth_header(token))
if response.status_code == 200:
return response.json()
def get_user():
"""Return user id if the token is valid, None otherwise."""
token = g.get(token, '')
response = api_get('sessions', token, {'token': token}, {'user': 1})
response = api_get('sessions', {'token': g.get('token', '')}, {'user': 1})
if response:
return response['_items'][0]['user']
......@@ -54,11 +72,10 @@ def get_user():
def get_nethz():
"""Return nethz of current user."""
if get_user() is not None:
response = api_get('users/%s' % get_user(), token, {}, {'nethz': 1})
response = api_get('users/%s' % get_user(), {}, {'nethz': 1})
return response.get('nethz')
def is_admin():
"""Return True if user is in the 'PVK Admins' Group, False otherwise.
......@@ -66,17 +83,16 @@ def is_admin():
The result is saved in g, to avoid checking twice, so there is no
performance loss if is_admin is called multiple times during a request.
token = g.get(token, '')
user_id = get_user()
if user_id is not None:
# Find Group with correct name, return list of members
groups = api_get('groups', token,
groups = api_get('groups',
{'name': current_app.config['ADMIN_GROUP_NAME']},
{'_id': 1})
if groups:
group_id = groups['_items'][0]['_id']
membership = api_get('groupmemberships', token,
membership = api_get('groupmemberships',
{'user': user_id, 'group': group_id},
{'_id': 1})
......@@ -86,11 +102,17 @@ def is_admin():
return False
# Auth
class APIAuth(TokenAuth):
"""Verifies the presented with AMIVAPI."""
"""Verifies the request token with AMIVAPI."""
def check_auth(self, token, allowed_roles, resource, method):
"""Allow request if token exists in AMIVAPI."""
"""Allow request if token exists in AMIVAPI.
Furthermore, grant admin rights if the user is member of the
admin group.
g.token = token # Safe in g such that other methods can use it
# Valid Login always required
......@@ -102,7 +124,7 @@ class APIAuth(TokenAuth):
return method == 'GET' or (resource == 'signups') or is_admin()
# Hook to filter signups
# Dynamic Filter
def only_own_signups(request, lookup):
"""Users can only see signups if their ID matches."""
......@@ -112,7 +134,7 @@ def only_own_signups(request, lookup):
lookup.setdefault('$and', []).append({'nethz': get_nethz()})
# Validator that allows to check nethz
# Validation
class APIValidator(Validator):
"""Provide a rule to check nethz of current user."""
......@@ -140,12 +162,13 @@ class APIValidator(Validator):
if key not in self.document.keys():
lookup[key] = original[key]
if, None, **lookup) is not None:
resource = self.resource
if, None, **lookup) is not None:
self._error(field, "value already exists in the database in " +
"combination with values for: %s" %
def _validate_not_patchable(self, enabled, field, value):
"""Inhibit patching of the field, copied from AMIVAPI."""
"""Inhibit patching of the field, also copied from AMIVAPI."""
if enabled and (request.method == 'PATCH'):
self._error(field, "this field can not be changed with PATCH")
"""Data Model and General Configuration of Eve.
Check out [the Eve docs for configuration](
if you are unsure about some of the settings.
# AMIVAPI URL and Admin Group
"""Test fixtures."""
"""Test Helpers.
Most importantly, set the test db credentials and provide an `app` fixture
which tests can use to get an app object with test client.
This fixture automatically ensures that the database is dropped after the test.
Furthermore the test client is modified a bit compared to the default client
to allow easier assertion of status_codes and make json handling easier.
import json
......@@ -13,8 +22,8 @@ TEST_SETTINGS = {
'MONGO_HOST': 'localhost',
'MONGO_PORT': 27017,
'MONGO_DBNAME': 'pvk_test',
'MONGO_USERNAME': 'pvkuser',
'MONGO_PASSWORD': 'pvkpass',
'MONGO_USERNAME': 'pvk_user',
'MONGO_PASSWORD': 'pvk_pass',
......@@ -54,5 +63,3 @@ def app():
application.client = application.test_client()
yield application
"""Tests for basic requests to all resources."""
"""Tests for basic requests to all resources as admin."""
from flask import g
"""Tests for authentication."""
"""Tests for all security function."""
import pytest
......@@ -27,7 +27,6 @@ def test_auth_required_for_item(app, resource, method):
@pytest.mark.parametrize('resource', ['lectures', 'assistants', 'courses'])
def test_user_can_read(app, resource):
"""Users should be able to to GET requests on resource and item.
......@@ -145,6 +144,7 @@ def test_user_signup_visibility(app):
app.client.delete(own_url, headers=token, assert_status=412)
app.client.delete(other_url, headers=token, assert_status=404)
def test_admin_signup_visibility(app):
"""Test that we an admin can see others' signups."""
with app.test_request_context():
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment