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
|
# 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/>.
from importlib import import_module
from typing import Any, Dict, Optional
from ariadne import ObjectType, convert_kwargs_to_snake_case, convert_camel_case_to_snake
from graphql import GraphQLResolveInfo
from makefun import with_signature
from .. import state
from .. libs import key_auth
from api.graphql.session.session import Session
from api.graphql.session.errors.op_mode_errors import op_mode_err_msg, op_mode_err_code
from vyos.opmode import Error as OpModeError
query = ObjectType("Query")
def make_query_resolver(query_name, class_name, session_func):
"""Dynamically generate a resolver for the query named in the
schema by 'query_name'.
Dynamic generation is provided using the package 'makefun' (via the
decorator 'with_signature'), which provides signature-preserving
function wrappers; it provides several improvements over, say,
functools.wraps.
:raise Exception:
raising ConfigErrors, or internal errors
"""
func_base_name = convert_camel_case_to_snake(class_name)
resolver_name = f'resolve_{func_base_name}'
func_sig = '(obj: Any, info: GraphQLResolveInfo, data: Optional[Dict]=None)'
@query.field(query_name)
@convert_kwargs_to_snake_case
@with_signature(func_sig, func_name=resolver_name)
async def func_impl(*args, **kwargs):
try:
auth_type = state.settings['app'].state.vyos_auth_type
if auth_type == 'key':
data = kwargs['data']
key = data['key']
auth = key_auth.auth_required(key)
if auth is None:
return {
"success": False,
"errors": ['invalid API key']
}
# We are finished with the 'key' entry, and may remove so as to
# pass the rest of data (if any) to function.
del data['key']
elif auth_type == 'token':
data = kwargs['data']
if data is None:
data = {}
info = kwargs['info']
user = info.context.get('user')
if user is None:
error = info.context.get('error')
if error is not None:
return {
"success": False,
"errors": [error]
}
return {
"success": False,
"errors": ['not authenticated']
}
else:
# AtrributeError will have already been raised if no
# vyos_auth_type; validation and defaultValue ensure it is
# one of the previous cases, so this is never reached.
pass
session = state.settings['app'].state.vyos_session
# one may override the session functions with a local subclass
try:
mod = import_module(f'api.graphql.session.override.{func_base_name}')
klass = getattr(mod, class_name)
except ImportError:
# otherwise, dynamically generate subclass to invoke subclass
# name based functions
klass = type(class_name, (Session,), {})
k = klass(session, data)
method = getattr(k, session_func)
result = method()
data['result'] = result
return {
"success": True,
"data": data
}
except OpModeError as e:
typename = type(e).__name__
msg = str(e)
return {
"success": False,
"errors": ['op_mode_error'],
"op_mode_error": {"name": f"{typename}",
"message": msg if msg else op_mode_err_msg.get(typename, "Unknown"),
"vyos_code": op_mode_err_code.get(typename, 9999)}
}
except Exception as error:
return {
"success": False,
"errors": [repr(error)]
}
return func_impl
def make_config_session_query_resolver(query_name):
return make_query_resolver(query_name, query_name,
convert_camel_case_to_snake(query_name))
def make_gen_op_query_resolver(query_name):
return make_query_resolver(query_name, query_name, 'gen_op_query')
def make_composite_query_resolver(query_name):
return make_query_resolver(query_name, query_name,
convert_camel_case_to_snake(query_name))
|
