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
|
"""User authorisation endpoints."""
import traceback
from typing import Tuple, Optional
import sqlite3
from flask import request, jsonify, Response, Blueprint, current_app
from gn3.auth import db
from gn3.auth.dictify import dictify
from ..groups.models import user_group as _user_group
from ..errors import NotFoundError, UserRegistrationError
from ..resources.models import user_resources as _user_resources
from ..roles.models import assign_default_roles, user_roles as _user_roles
from ...authentication.oauth2.resource_server import require_oauth
from ...authentication.users import save_user, set_user_password
from ...authentication.oauth2.models.oauth2token import token_by_access_token
users = Blueprint("users", __name__)
@users.route("/", methods=["GET"])
@require_oauth("profile")
def user_details() -> Response:
"""Return user's details."""
with require_oauth.acquire("profile") as the_token:
user = the_token.user
user_dets = {
"user_id": user.user_id, "email": user.email, "name": user.name,
"group": False
}
with db.connection(current_app.config["AUTH_DB"]) as conn, db.cursor(conn) as cursor:
the_group = _user_group(cursor, user).maybe(# type: ignore[misc]
False, lambda grp: grp)# type: ignore[arg-type]
return jsonify({
**user_dets,
"group": dictify(the_group) if the_group else False
})
@users.route("/roles", methods=["GET"])
@require_oauth("role")
def user_roles() -> Response:
"""Return the non-resource roles assigned to the user."""
with require_oauth.acquire("role") as token:
with db.connection(current_app.config["AUTH_DB"]) as conn:
return jsonify(tuple(
dictify(role) for role in
_user_roles(conn, token.user).maybe(# type: ignore[misc]
tuple(), lambda roles: roles)))
def __email_valid__(email: str) -> Tuple[bool, Optional[str]]:
"""Validate the email address."""
if email == "":
return False, "Empty email address"
## Check that the address is a valid email address
## Review use of `email-validator` or `pyIsEmail` python packages for
## validating the emails, if it turns out this is important.
## Success
return True, None
def __password_valid__(password, confirm_password) -> Tuple[bool, Optional[str]]:
if password == "" or confirm_password == "":
return False, "Empty password value"
if password != confirm_password:
return False, "Mismatched password values"
return True, None
def __user_name_valid__(name: str) -> Tuple[bool, Optional[str]]:
if name == "":
return False, "User's name not provided."
return True, None
def __assert_not_logged_in__(conn: db.DbConnection):
bearer = request.headers.get('Authorization')
if bearer:
token = token_by_access_token(conn, bearer.split(None)[1]).maybe(# type: ignore[misc]
False, lambda tok: tok)
if token:
raise UserRegistrationError(
"Cannot register user while authenticated")
@users.route("/register", methods=["POST"])
def register_user() -> Response:
"""Register a user."""
with db.connection(current_app.config["AUTH_DB"]) as conn:
__assert_not_logged_in__(conn)
form = request.form
email = form.get("email", "").strip()
password = form.get("password", "").strip()
user_name = form.get("user_name", "").strip()
errors = tuple(
error for valid,error in
[__email_valid__(email),
__password_valid__(
password, form.get("confirm_password", "").strip()),
__user_name_valid__(user_name)]
if not valid)
if len(errors) > 0:
raise UserRegistrationError(*errors)
try:
with db.cursor(conn) as cursor:
user, _hashed_password = set_user_password(
cursor, save_user(cursor, email, user_name), password)
assign_default_roles(cursor, user)
return jsonify(
{
"user_id": user.user_id,
"email": user.email,
"name": user.name
})
except sqlite3.IntegrityError as sq3ie:
current_app.logger.debug(traceback.format_exc())
raise UserRegistrationError(
"A user with that email already exists") from sq3ie
raise Exception(
"unknown_error", "The system experienced an unexpected error.")
@users.route("/group", methods=["GET"])
@require_oauth("profile group")
def user_group() -> Response:
"""Retrieve the group in which the user is a member."""
with require_oauth.acquire("profile group") as the_token:
db_uri = current_app.config["AUTH_DB"]
with db.connection(db_uri) as conn, db.cursor(conn) as cursor:
group = _user_group(cursor, the_token.user).maybe(# type: ignore[misc]
False, lambda grp: grp)# type: ignore[arg-type]
if group:
return jsonify(dictify(group))
raise NotFoundError("User is not a member of any group.")
@users.route("/resources")
@require_oauth("profile resource")
def user_resources() -> Response:
"""Retrieve the resources a user has access to."""
with require_oauth.acquire("profile resource") as the_token:
db_uri = current_app.config["AUTH_DB"]
with db.connection(db_uri) as conn:
return jsonify([
dictify(resource) for resource in
_user_resources(conn, the_token.user)])
|
