diff --git a/src/services/api/graphql/graphql/auth_token_mutation.py b/src/services/api/graphql/graphql/auth_token_mutation.py
index 21ac40094..603a13758 100644
--- a/src/services/api/graphql/graphql/auth_token_mutation.py
+++ b/src/services/api/graphql/graphql/auth_token_mutation.py
@@ -1,49 +1,61 @@
# Copyright 2022 VyOS maintainers and contributors <maintainers@vyos.io>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this library. If not, see <http://www.gnu.org/licenses/>.
import jwt
import datetime
from typing import Any, Dict
from ariadne import ObjectType, UnionType
from graphql import GraphQLResolveInfo
from .. libs.token_auth import generate_token
+from .. session.session import get_user_info
from .. import state
auth_token_mutation = ObjectType("Mutation")
@auth_token_mutation.field('AuthToken')
def auth_token_resolver(obj: Any, info: GraphQLResolveInfo, data: Dict):
# non-nullable fields
user = data['username']
passwd = data['password']
secret = state.settings['secret']
exp_interval = int(state.settings['app'].state.vyos_token_exp)
expiration = (datetime.datetime.now(tz=datetime.timezone.utc) +
datetime.timedelta(seconds=exp_interval))
res = generate_token(user, passwd, secret, expiration)
- if res:
+ try:
+ res |= get_user_info(user)
+ except ValueError:
+ # non-existent user already caught
+ pass
+ if 'token' in res:
data['result'] = res
return {
"success": True,
"data": data
}
+ if 'errors' in res:
+ return {
+ "success": False,
+ "errors": res['errors']
+ }
+
return {
"success": False,
"errors": ['token generation failed']
}
diff --git a/src/services/api/graphql/libs/token_auth.py b/src/services/api/graphql/libs/token_auth.py
index 2100eba7f..8585485c9 100644
--- a/src/services/api/graphql/libs/token_auth.py
+++ b/src/services/api/graphql/libs/token_auth.py
@@ -1,71 +1,70 @@
import jwt
import uuid
import pam
from secrets import token_hex
from .. import state
def _check_passwd_pam(username: str, passwd: str) -> bool:
if pam.authenticate(username, passwd):
return True
return False
def init_secret():
length = int(state.settings['app'].state.vyos_secret_len)
secret = token_hex(length)
state.settings['secret'] = secret
def generate_token(user: str, passwd: str, secret: str, exp: int) -> dict:
if user is None or passwd is None:
return {}
if _check_passwd_pam(user, passwd):
app = state.settings['app']
try:
users = app.state.vyos_token_users
except AttributeError:
app.state.vyos_token_users = {}
users = app.state.vyos_token_users
user_id = uuid.uuid1().hex
payload_data = {'iss': user, 'sub': user_id, 'exp': exp}
secret = state.settings.get('secret')
if secret is None:
- return {
- "success": False,
- "errors": ['failed secret generation']
- }
+ return {"errors": ['missing secret']}
token = jwt.encode(payload=payload_data, key=secret, algorithm="HS256")
users |= {user_id: user}
return {'token': token}
+ else:
+ return {"errors": ['failed pam authentication']}
def get_user_context(request):
context = {}
context['request'] = request
context['user'] = None
if 'Authorization' in request.headers:
auth = request.headers['Authorization']
scheme, token = auth.split()
if scheme.lower() != 'bearer':
return context
try:
secret = state.settings.get('secret')
payload = jwt.decode(token, secret, algorithms=["HS256"])
user_id: str = payload.get('sub')
if user_id is None:
return context
except jwt.exceptions.ExpiredSignatureError:
context['error'] = 'expired token'
return context
except jwt.PyJWTError:
return context
try:
users = state.settings['app'].state.vyos_token_users
except AttributeError:
return context
user = users.get(user_id)
if user is not None:
context['user'] = user
return context
diff --git a/src/services/api/graphql/session/session.py b/src/services/api/graphql/session/session.py
index b2aef9bd9..3c5a062b6 100644
--- a/src/services/api/graphql/session/session.py
+++ b/src/services/api/graphql/session/session.py
@@ -1,204 +1,212 @@
# Copyright 2021-2022 VyOS maintainers and contributors <maintainers@vyos.io>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this library. If not, see <http://www.gnu.org/licenses/>.
import os
import json
from ariadne import convert_camel_case_to_snake
from vyos.config import Config
from vyos.configtree import ConfigTree
from vyos.defaults import directories
from vyos.template import render
from vyos.opmode import Error as OpModeError
from api.graphql.libs.op_mode import load_op_mode_as_module, split_compound_op_mode_name
from api.graphql.libs.op_mode import normalize_output
op_mode_include_file = os.path.join(directories['data'], 'op-mode-standardized.json')
+def get_config_dict(path=[], effective=False, key_mangling=None,
+ get_first_key=False, no_multi_convert=False,
+ no_tag_node_value_mangle=False):
+ config = Config()
+ return config.get_config_dict(path=path, effective=effective,
+ key_mangling=key_mangling,
+ get_first_key=get_first_key,
+ no_multi_convert=no_multi_convert,
+ no_tag_node_value_mangle=no_tag_node_value_mangle)
+
+def get_user_info(user):
+ user_info = {}
+ info = get_config_dict(['system', 'login', 'user', user],
+ get_first_key=True)
+ if not info:
+ raise ValueError("No such user")
+
+ user_info['user'] = user
+ user_info['full_name'] = info.get('full-name', '')
+
+ return user_info
+
class Session:
"""
Wrapper for calling configsession functions based on GraphQL requests.
Non-nullable fields in the respective schema allow avoiding a key check
in 'data'.
"""
def __init__(self, session, data):
self._session = session
self._data = data
self._name = convert_camel_case_to_snake(type(self).__name__)
try:
with open(op_mode_include_file) as f:
self._op_mode_list = json.loads(f.read())
except Exception:
self._op_mode_list = None
- @staticmethod
- def _get_config_dict(path=[], effective=False, key_mangling=None,
- get_first_key=False, no_multi_convert=False,
- no_tag_node_value_mangle=False):
- config = Config()
- return config.get_config_dict(path=path, effective=effective,
- key_mangling=key_mangling,
- get_first_key=get_first_key,
- no_multi_convert=no_multi_convert,
- no_tag_node_value_mangle=no_tag_node_value_mangle)
-
def show_config(self):
session = self._session
data = self._data
out = ''
try:
out = session.show_config(data['path'])
if data.get('config_format', '') == 'json':
config_tree = ConfigTree(out)
out = json.loads(config_tree.to_json())
except Exception as error:
raise error
return out
def save_config_file(self):
session = self._session
data = self._data
if 'file_name' not in data or not data['file_name']:
data['file_name'] = '/config/config.boot'
try:
session.save_config(data['file_name'])
except Exception as error:
raise error
def load_config_file(self):
session = self._session
data = self._data
try:
session.load_config(data['file_name'])
session.commit()
except Exception as error:
raise error
def show(self):
session = self._session
data = self._data
out = ''
try:
out = session.show(data['path'])
except Exception as error:
raise error
return out
def add_system_image(self):
session = self._session
data = self._data
try:
res = session.install_image(data['location'])
except Exception as error:
raise error
return res
def delete_system_image(self):
session = self._session
data = self._data
try:
res = session.remove_image(data['name'])
except Exception as error:
raise error
return res
def show_user_info(self):
session = self._session
data = self._data
user_info = {}
user = data['user']
try:
- info = self._get_config_dict(['system', 'login', 'user', user,
- 'full-name'])
- user_info['user'] = user
- user_info['full_name'] = info.get('full-name', '')
+ user_info = get_user_info(user)
except Exception as error:
raise error
return user_info
def system_status(self):
import api.graphql.session.composite.system_status as system_status
session = self._session
data = self._data
status = {}
status['host_name'] = session.show(['host', 'name']).strip()
status['version'] = system_status.get_system_version()
status['uptime'] = system_status.get_system_uptime()
status['ram'] = system_status.get_system_ram_usage()
return status
def gen_op_query(self):
session = self._session
data = self._data
name = self._name
op_mode_list = self._op_mode_list
# handle the case that the op-mode file contains underscores:
if op_mode_list is None:
raise FileNotFoundError(f"No op-mode file list at '{op_mode_include_file}'")
(func_name, scriptname) = split_compound_op_mode_name(name, op_mode_list)
if scriptname == '':
raise FileNotFoundError(f"No op-mode file named in string '{name}'")
mod = load_op_mode_as_module(f'{scriptname}')
func = getattr(mod, func_name)
try:
res = func(True, **data)
except OpModeError as e:
raise e
res = normalize_output(res)
return res
def gen_op_mutation(self):
session = self._session
data = self._data
name = self._name
op_mode_list = self._op_mode_list
# handle the case that the op-mode file name contains underscores:
if op_mode_list is None:
raise FileNotFoundError(f"No op-mode file list at '{op_mode_include_file}'")
(func_name, scriptname) = split_compound_op_mode_name(name, op_mode_list)
if scriptname == '':
raise FileNotFoundError(f"No op-mode file named in string '{name}'")
mod = load_op_mode_as_module(f'{scriptname}')
func = getattr(mod, func_name)
try:
res = func(**data)
except OpModeError as e:
raise e
return res