kepconfig.ua_gateway.server

  1# -------------------------------------------------------------------------
  2# Copyright (c) 2023 PTC Inc. and/or all its affiliates. All rights reserved.
  3# See License.txt in the project root for
  4# license information.
  5# --------------------------------------------------------------------------
  6
  7r"""`server` exposes an API to allow modifications (add, delete, modify) to 
  8UA Gateway plug-in server endpoint objects within the Kepware Configuration API. 
  9Certificate store read, remove and trust functionality is also available for the 
 10server endpoints.
 11"""
 12
 13from typing import Union
 14
 15from kepconfig.structures import KepServiceResponse
 16from ..connection import server
 17from ..error import KepError, KepHTTPError
 18from ..ua_gateway.common import _INTER_TYPE, _change_cert_trust, _create_url_cert, _create_url_server, _delete_cert_truststore, SERVER_ROOT, _create_url_inst_cert
 19
 20SERVER_INSTANCE_CERTIFICATE = 'Server Instance Certificate'
 21
 22def get_uag_server_interface_properties(server: server) -> dict:
 23    ''' Get the UAG Server Interface Properties of the Kepware instance. These properties expose User Identify
 24    Policy, Security Policy and other Communication properties that are applied across all UAG server endpoints.
 25        
 26    :return: Dict of all the UAG Server Interface properties
 27
 28    :raises KepHTTPError: If urllib provides an HTTPError
 29    :raises KepURLError: If urllib provides an URLError
 30    '''
 31
 32    r = server._config_get(server.url + SERVER_ROOT)
 33    return r.payload
 34
 35def modify_uag_server_interface_properties(server: server, DATA: dict, force: bool = False) -> bool:
 36    ''' Modify the UAG Server Interface Properties of the Kepware instance. These properties expose User Identify
 37    Policy, Security Policy and other Communication properties that are applied across all UAG server endpoints.
 38
 39    :param DATA: Dict of the UAG Server Interface properties to be modified
 40    :param force: *(optional)* if True, will force the configuration update to the Kepware server
 41    
 42    :return: True - If a "HTTP 200 - OK" is received from Kepware server
 43
 44    :raises KepHTTPError: If urllib provides an HTTPError
 45    :raises KepURLError: If urllib provides an URLError
 46    '''
 47
 48    prop_data = server._force_update_check(force, DATA)
 49    r = server._config_update(server.url + SERVER_ROOT, prop_data)
 50    if r.code == 200: return True 
 51    else: raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
 52
 53
 54def get_ua_server_endpoint(server: server, ua_server_endpoint: str) -> dict:
 55    '''Returns the properties of the UAG server endpoint object.
 56    
 57    :param server: instance of the `server` class
 58    :param ua_server_endpoint: name of ua_server_endpoint
 59    
 60    :return: Dict of data for the server endpoint requested
 61
 62    :raises KepHTTPError: If urllib provides an HTTPError
 63    :raises KepURLError: If urllib provides an URLError
 64    '''
 65    r = server._config_get(server.url + _create_url_server(ua_server_endpoint))
 66    return r.payload
 67
 68def get_all_ua_server_endpoints(server: server,  *, options: dict = None) -> dict:
 69    '''Returns list of all UAG server endpoint objects and their properties.
 70    
 71    :param server: instance of the `server` class
 72    :param options: *(optional)* Dict of parameters to filter, sort or pagenate the list of server endpoints. Options are `filter`, 
 73        `sortOrder`, `sortProperty`, `pageNumber`, and `pageSize`
 74    
 75    :return: List of data for all server endpoints in Kepware server UAG
 76
 77    :raises KepHTTPError: If urllib provides an HTTPError
 78    :raises KepURLError: If urllib provides an URLError
 79    '''
 80    r = server._config_get(server.url + _create_url_server())
 81    return r.payload
 82
 83def add_ua_server_endpoint(server: server, DATA: Union[dict, list]) -> Union[bool, list]:
 84    '''Add a `"UAG server endpoint"` or multiple `"UAG server endpoint"` objects to Kepware. This allows you 
 85    to create a server endpoint with all needed properties.
 86
 87    Additionally it can be used to pass a list of UAG server endpoint to be added all at once.
 88
 89    :param server: instance of the `server` class
 90    :param DATA: Dict of the endpoint or a list of endpoints
 91    expected by Kepware Configuration API
 92
 93    :return: True - If a "HTTP 201 - Created" is received from Kepware server
 94    :return: If a "HTTP 207 - Multi-Status" is received from Kepware with a list of dict error responses for all 
 95    endpoints added that failed.
 96
 97    :raises KepHTTPError: If urllib provides an HTTPError
 98    :raises KepURLError: If urllib provides an URLError
 99    '''
100    r = server._config_add(server.url + _create_url_server(), DATA)
101    if r.code == 201: return True
102    elif r.code == 207:
103        errors = [] 
104        for item in r.payload:
105            if item['code'] != 201:
106                errors.append(item)
107        return errors
108    else: 
109        raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
110    
111def modify_ua_server_endpoint(server: server, DATA: dict, *, ua_server_endpoint: str = None, force: bool = False) -> bool:
112    '''Modify a UAG server endpoint object and it's properties in Kepware. If a `"ua_server_endpoint"` is not provided as an input,
113    you need to identify the client connection in the *'common.ALLTYPES_NAME'* property field in `"DATA"`. It will 
114    assume that is the client connection that is to be modified.
115
116    :param server: instance of the `server` class
117    :param DATA: Dict of the `ua_server_endpoint` properties to be modified
118    :param ua_server_endpoint: *(optional)* name of server endpoint to modify. Only needed if not existing in `"DATA"`
119    :param force: *(optional)* if True, will force the configuration update to the Kepware server
120    
121    :return: True - If a "HTTP 200 - OK" is received from Kepware server
122
123    :raises KepHTTPError: If urllib provides an HTTPError
124    :raises KepURLError: If urllib provides an URLError
125    '''
126
127    server_data = server._force_update_check(force, DATA)
128    if ua_server_endpoint == None:
129        try:
130            r = server._config_update(server.url + _create_url_server(server_data['common.ALLTYPES_NAME']), server_data)
131            if r.code == 200: return True 
132            else: raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
133        except KeyError as err:
134            err_msg = 'Error: No Channel identified in DATA | Key Error: {}'.format(err)
135            raise KepError(err_msg)
136        # except Exception as e:
137        #     return 'Error: Error with {}: {}'.format(inspect.currentframe().f_code.co_name, str(e))
138    else:
139        r = server._config_update(server.url + _create_url_server(ua_server_endpoint), server_data)
140        if r.code == 200: return True 
141        else: 
142            raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
143        
144def del_ua_server_endpoint(server: server, ua_server_endpoint: str) -> bool:
145    '''Delete a `"UAG server endpoint"` object in Kepware.
146    
147    :param server: instance of the `server` class
148    :param ua_server_endpoint: name of server endpoint
149    
150    :return: True - If a "HTTP 200 - OK" is received from Kepware server
151
152    :raises KepHTTPError: If urllib provides an HTTPError
153    :raises KepURLError: If urllib provides an URLError
154    '''
155    r = server._config_del(server.url + _create_url_server(ua_server_endpoint))
156    if r.code == 200: return True 
157    else: 
158        raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
159
160def get_certificate(server: server, certificate: str) -> dict:
161    '''Returns the properties of the UAG server endpoint certificate object in the UAG client connection 
162    certificate store. These are UA client instance certificates that are used by UAG server endpoints for
163    trust purposes in the UA security model.
164    
165    :param server: instance of the `server` class
166    :param certificate: name of certificate
167    
168    :return: Dict of data for the certificate requested
169
170    :raises KepHTTPError: If urllib provides an HTTPError
171    :raises KepURLError: If urllib provides an URLError
172    '''
173    r = server._config_get(server.url + _create_url_cert(_INTER_TYPE.SERVER, certificate))
174    return r.payload
175
176def get_all_certificates(server: server, *, options: dict = None) -> list:
177    '''Returns list of all UAG server endpoint certificate objects and their properties.This is updating the trust state of UA client instance 
178    certificates that are used by UAG server endpoints for trust purposes in the UA security model. These are UA client instance certificates 
179    that are used by UAG server endpoints for trust purposes in the UA security model.
180    
181    :param server: instance of the `server` class
182    :param options: *(optional)* Dict of parameters to filter, sort or pagenate the list of certificates. Options are `filter`, 
183        `sortOrder`, `sortProperty`, `pageNumber`, and `pageSize`
184    
185    :return: List of data for all certificates in Kepware server UAG server endpoint certificate store
186
187    :raises KepHTTPError: If urllib provides an HTTPError
188    :raises KepURLError: If urllib provides an URLError
189    '''
190    r = server._config_get(server.url + _create_url_cert(_INTER_TYPE.SERVER), params= options)
191    return r.payload
192
193def trust_certificate(server: server, certificate: str) -> bool:
194    '''Trusts the certificate in the UAG server endpoint certificate store. This is updating the trust state of UA client instance 
195    certificates that are used by UAG server endpoints for trust purposes in the UA security model.
196
197    :param server: instance of the `server` class
198    :param certificate: name of certificate
199    
200    :return: True - If a "HTTP 200 - OK" is received from Kepware server
201
202    :raises KepHTTPError: If urllib provides an HTTPError
203    :raises KepURLError: If urllib provides an URLError
204    '''
205    return _change_cert_trust(server, _INTER_TYPE.SERVER, certificate, True)
206
207def reject_certificate(server: server, certificate: str) -> bool:
208    '''Rejects the certificate in the UAG server endpoint certificate store.
209
210    :param server: instance of the `server` class
211    :param certificate: name of certificate
212    
213    :return: True - If a "HTTP 200 - OK" is received from Kepware server
214
215    :raises KepHTTPError: If urllib provides an HTTPError
216    :raises KepURLError: If urllib provides an URLError
217    '''
218    return _change_cert_trust(server, _INTER_TYPE.SERVER, certificate, False)
219
220def delete_certificate(server: server, certificate: str) -> bool:
221    '''Deletes the certificate in the UAG server endpoint certificate store.
222
223    :param server: instance of the `server` class
224    :param certificate: name of certificate
225    
226    :return: True - If a "HTTP 200 - OK" is received from Kepware server
227
228    :raises KepHTTPError: If urllib provides an HTTPError
229    :raises KepURLError: If urllib provides an URLError
230    '''
231    return _delete_cert_truststore(server, _INTER_TYPE.SERVER, certificate)
232
233def get_instance_certificate(server: server) -> dict:
234    '''Returns the properties of the UAG server instance certificate object in the UAG certificate store. 
235    These are UAG instance certificates that are used by UAG for trust purposes in the UA security model.
236    
237    :param server: instance of the `server` class
238    
239    :return: Dict of properties for the certificate requested
240
241    :raises KepHTTPError: If urllib provides an HTTPError
242    :raises KepURLError: If urllib provides an URLError
243    '''
244    r = server._config_get(server.url + _create_url_inst_cert(_INTER_TYPE.SERVER, SERVER_INSTANCE_CERTIFICATE))
245    return r.payload
246
247def reissue_self_signed_instance_certificate(server: server, job_ttl: int = None) -> KepServiceResponse:
248    '''Deletes and reissues a self-signed UAG server instance certificate object in the UAG certificate store. 
249    This is the UAG instance certificate that are used by UAG for trust purposes in the UA security model.
250    
251    :param server: instance of the `server` class
252    :param job_ttl: *(optional)* Determines the number of seconds a job instance will exist following completion.
253
254    :return: `KepServiceResponse` instance with job information
255
256    :raises KepHTTPError: If urllib provides an HTTPError
257    :raises KepURLError: If urllib provides an URLError
258    '''
259    url = server.url + _create_url_inst_cert(_INTER_TYPE.SERVER, SERVER_INSTANCE_CERTIFICATE) + '/services/ReIssueInstanceCertificate'
260    try:
261        job = server._kep_service_execute(url, TTL= job_ttl)
262        return job
263    except Exception as err:
264        raise err