kepconfig.connection

  1# -------------------------------------------------------------------------
  2# Copyright (c) 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"""`connection` exposes an `server` class that manages the connection
  8information and RESTful requests for the Kepware Configuration API Library.
  9"""
 10
 11import json
 12import codecs
 13import datetime
 14from urllib import request, parse, error
 15from base64 import b64encode
 16from .error import KepError, KepHTTPError, KepURLError
 17import socket
 18import ssl
 19from .structures import KepServiceResponse, KepServiceStatus, _HttpDataAbstract, Filter
 20
 21
 22class server:
 23    '''A class to represent a connection to an instance of Kepware. This object is used to 
 24    create the Authentication and server parameters to taget a Kepware instance. An instance of this is 
 25    used in all configuration calls done.
 26
 27    :param host: host name or IP address
 28    :param port: port of Configuration API
 29    :param username: username to conduct "Basic Authentication"
 30    :param password: password to conduct "Basic Authentication"
 31    :param https: Sets `SSL_on` to use HTTPS connection (Default: False)
 32    :param SSL_on: Identify to use HTTPS connection (Default: False)
 33    :param SSL_ignore_hostname: During certificate validation ignore the hostname check
 34    :param SSL_trust_all_certs: (insecure) - During certificate validation trust any certificate - if True, 
 35        will "set SSL_ignore_hostname" to true
 36    :param url: base URL for the server connection
 37
 38    **Methods**
 39
 40    :meth:`reinitialize`: reinitialize the Kepware server
 41
 42    :meth:`get_transaction_log`: retrieve the Configuration API transaction logs
 43
 44    :meth:`get_event_log`: retrieve the Kepware Event Log
 45
 46    :meth:`get_audit_log`: retrieve the Kepware Audit Log
 47
 48    :meth:`get_info`: retrieve the Kepware product information
 49    
 50    :meth:`import_empty_project`: import an empty project to the Kepware server
 51
 52    :meth:`get_project_properties`: retrieve the Kepware Project Properties
 53
 54    :meth:`modify_project_properties` - modify the Kepware Project Properties
 55
 56    :meth:`service_status` - retrive service job status
 57
 58    :meth:`export_project_configuration` - export the current project configuration in JSON format
 59
 60    :meth:`save_project` - save the current project to a file
 61
 62    :meth:`load_project` - load a project from a file
 63    '''
 64    __root_url = '/config'
 65    __version_url = '/v1'
 66    __project_services_url = '/project/services'
 67    __event_log_url = '/event_log'
 68    __trans_log_url = '/log'
 69    __audit_log_url = '/audit_log'
 70
 71
 72
 73    def __init__(self,  host: str, port: int, user: str, pw: str, https: bool = False):
 74        self.host = host
 75        self.port = port
 76        self.username = user
 77        self.password = pw
 78        self.__ssl_context = ssl.create_default_context()
 79        self.__SSL_on = https
 80    
 81    @property
 82    def url(self):
 83        if self.SSL_on:
 84            proto = 'https'
 85        else:
 86            proto = 'http'
 87        return  f'{proto}://{self.host}:{self.port}{self.__root_url}{self.__version_url}'
 88    
 89    @property
 90    def SSL_on(self):
 91        return self.__SSL_on
 92    
 93    @SSL_on.setter
 94    def SSL_on(self, val):
 95        
 96        if isinstance(val, bool):
 97            self.__SSL_on = val
 98
 99    @property
100    def SSL_ignore_hostname(self):
101        return not self.__ssl_context.check_hostname
102
103    @SSL_ignore_hostname.setter
104    def SSL_ignore_hostname(self, val):
105        if isinstance(val, bool):
106            if val == True:
107                self.__ssl_context.check_hostname = False
108            else:
109                self.__ssl_context.check_hostname = True
110
111    
112    @property
113    def SSL_trust_all_certs(self):
114        if self.__ssl_context.verify_mode == ssl.CERT_NONE:
115            return True
116        else:
117            return False
118
119    @SSL_trust_all_certs.setter
120    def SSL_trust_all_certs(self, val):
121        if isinstance(val, bool):
122            if val == True:
123                if self.__ssl_context.check_hostname == True:
124                    self.__ssl_context.check_hostname = False
125                self.__ssl_context.verify_mode = ssl.CERT_NONE
126            else:
127                self.__ssl_context.verify_mode = ssl.CERT_REQUIRED
128
129
130    def get_status(self) -> dict:
131        '''Executes a health status request to the Kepware instance to report service statuses.
132
133        :return: List of data for the health status request
134
135        :raises KepHTTPError: If urllib provides an HTTPError
136        :raises KepURLError: If urllib provides an URLError
137        '''
138
139        r = self._config_get(f'{self.url}/status')
140        return r.payload
141    def get_info(self) -> dict:
142        '''Requests product information from the Kepware instance. Provides various information including
143        product name and version information.
144
145        :return: dict of data for the product information request
146        
147        :raises KepHTTPError: If urllib provides an HTTPError
148        :raises KepURLError: If urllib provides an URLError
149        '''
150        
151        r = self._config_get(f'{self.url}/about')
152        return r.payload
153
154    def reinitialize(self, job_ttl: int = None) -> KepServiceResponse:
155        '''Executes a Reinitialize service call to the Kepware instance.
156
157        :param job_ttl: *(optional)* Determines the number of seconds a job instance will exist following completion.
158
159        :return: `KepServiceResponse` instance with job information
160
161        :raises KepHTTPError: If urllib provides an HTTPError (If not HTTP code 202 [Accepted] or 429 [Too Busy] returned)
162        :raises KepURLError: If urllib provides an URLError
163        '''
164        url = self.url + self.__project_services_url + '/ReinitializeRuntime' 
165        try:
166            job = self._kep_service_execute(url, None, job_ttl)
167            return job
168        except Exception as err:
169            raise err
170        
171    def get_transaction_log(self, limit: int = None, start: datetime.datetime = None, end: datetime.datetime = None) -> list:
172        ''' Get the Transaction Log from the Kepware instance.
173
174        :param limit: *(optional)* number of transaction log entries to request
175        :param start: *(optional)* start time of query as `datetime.datetime` type and should be UTC
176        :param end: *(optional)* end time of query as `datetime.datetime` type and should be UTC
177
178        :raises KepHTTPError: If urllib provides an HTTPError
179        :raises KepURLError: If urllib provides an URLError
180        '''
181        query = self.__create_query(start, end, limit)
182        url = f'{self.url}{self.__trans_log_url}'
183        r = self._config_get(url, params= query)
184        return r.payload
185
186    def get_event_log(self, limit: int = None, start: datetime.datetime = None, end: datetime.datetime = None, *, options: dict = None) -> list:
187        ''' Get the Event Log from the Kepware instance.
188
189        :param limit: *(optional)* number of event log entries to request
190        :param start: *(optional)* start time of query as `datetime.datetime` type and should be UTC
191        :param end: *(optional)* end time of query as `datetime.datetime` type and should be UTC
192        :param options: *(optional)* Dict of parameters to filter, sort or pagenate the list of transactions. Options are `event`, 
193        `sortOrder`, `sortProperty`, `pageNumber`, and `pageSize`
194
195        :raises KepHTTPError: If urllib provides an HTTPError
196        :raises KepURLError: If urllib provides an URLError
197        '''
198        query = self.__create_query(start, end, limit)
199        if options is not None:
200            query = {**query, **options}
201        url = f'{self.url}{self.__event_log_url}'
202        r = self._config_get(url, params= query)
203        return r.payload
204
205    def get_audit_log(self, limit: int = None, *, filters: list[Filter] = None, options: dict = None) -> list:
206        ''' Get the Audit Log from the Kepware instance.
207
208        :param limit: *(optional)* number of event log entries to request
209        :param filters: *(optional)* list of filters that are used to control results returned from the log
210        :param options: *(optional)* Dict of parameters to filter, sort or pagenate the list of transactions. Options are `sortOrder`, 
211        `sortProperty`, `pageNumber`, and `pageSize`
212
213        :raises KepHTTPError: If urllib provides an HTTPError
214        :raises KepURLError: If urllib provides an URLError
215        '''
216        query = self.__create_filter_query(filters)
217        if limit is not None:
218            query['limit'] = limit
219        if options is not None:
220            query = {**query, **options}
221        url = f'{self.url}{self.__audit_log_url}'
222        r = self._config_get(url, params= query)
223        return r.payload
224    
225    def get_project_properties(self) -> dict:
226        ''' Get the Project Properties of the Kepware instance.
227        
228        :return: Dict of all the project properties
229
230        :raises KepHTTPError: If urllib provides an HTTPError
231        :raises KepURLError: If urllib provides an URLError
232        '''
233
234        r = self._config_get(self.url + '/project')
235        return r.payload
236    
237    def modify_project_properties(self, DATA: dict, force: bool = False) -> bool:
238        ''' Modify the Project Properties of the Kepware instance.
239
240        :param DATA: Dict of the project properties to be modified
241        :param force: *(optional)* if True, will force the configuration update to the Kepware server
242        
243        :return: True - If a "HTTP 200 - OK" is received from Kepware server
244
245        :raises KepHTTPError: If urllib provides an HTTPError
246        :raises KepURLError: If urllib provides an URLError
247        '''
248
249        prop_data = self._force_update_check(force, DATA)
250        r = self._config_update(self.url + '/project', prop_data)
251        if r.code == 200: return True 
252        else: raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
253    
254    def import_empty_project(self) -> KepServiceResponse:
255        '''Executes JsonProjectLoad Service call to the Kepware instance with an empty project. This service 
256        imports an empty project configuration, acts like a FILE->NEW action and 
257        stop communications while the new project replaces the current project in the Kepware runtime. 
258
259        :return: `KepServiceResponse` instance with job information
260        
261        :raises KepHTTPError: If urllib provides an HTTPError (If not HTTP code 202 [Accepted] or 429 [Too Busy] returned)
262        :raises KepURLError: If urllib provides an URLError
263        '''
264        return self.import_project_configuration({"project":{}})
265
266
267    def import_project_configuration(self, DATA: dict) -> KepServiceResponse:
268        '''Executes JsonProjectLoad Service call to the Kepware instance. This service imports project configuration 
269        data, expecting a complete project file in JSON/dict format. This service acts like a FILE->OPEN action and 
270        stop communications while the new project replaces the current project in the Kepware runtime. 
271    
272        :param DATA: Complete project configuration data in JSON/dict format. 
273
274        :return: `KepServiceResponse` instance with job information
275        
276        :raises KepHTTPError: If urllib provides an HTTPError (If not HTTP code 202 [Accepted] or 429 [Too Busy] returned)
277        :raises KepURLError: If urllib provides an URLError
278        '''
279        url = self.url + self.__project_services_url + '/JsonProjectLoad'
280        try:
281            job = self._kep_service_execute(url, DATA)
282            return job
283        except Exception as err:
284            raise err
285        
286    def export_project_configuration(self) -> dict:
287        '''Get a complete copy of the project configuration in JSON format. This will include the same 
288        configuration that is stored when you save the project file manually.
289
290        :return: Dict of the complete project configuration
291
292        :raises KepHTTPError: If urllib provides an HTTPError
293        :raises KepURLError: If urllib provides an URLError
294        '''
295        r = self._config_get(self.url + '/project', params= {"content": "serialize"})
296        return r.payload
297    
298    def save_project(self, filename: str, password: str = None, job_ttl: int = None) -> KepServiceResponse:
299        '''Executes a ProjectSave Service call to the Kepware instance. This saves 
300        a copy of the current project file to disk. The filename
301
302        :param filename: Relative file path and name of project file including the file extension to save.
303        Location of relative project file paths:
304         
305                TKS or KEP (Windows): C:\\PROGRAMDATA\\PTC\\Thingworx Kepware Server\\V6 or 
306                                C:\\PROGRAMDATA\\Kepware\\KEPServerEX\\V6
307                TKE (Linux):    /opt/tkedge/v1/user_data
308
309
310        :param password: *(optional)* Specify a password with which to  save an encrypted project file with.  
311            This password will be required to load this project file.
312        :param job_ttl: *(optional)* Determines the number of seconds a job instance will exist following completion.
313
314        :return: `KepServiceResponse` instance with job information
315
316        :raises KepHTTPError: If urllib provides an HTTPError (If not HTTP code 202 [Accepted] or 429 [Too Busy] returned)
317        :raises KepURLError: If urllib provides an URLError
318        '''
319        url = self.url + self.__project_services_url + '/ProjectSave'
320        prop_data = {'servermain.PROJECT_FILENAME': filename}
321        if password != None: prop_data['servermain.PROJECT_PASSWORD'] = password
322        try:
323            job = self._kep_service_execute(url, prop_data, job_ttl)
324            return job
325        except Exception as err:
326            raise err
327
328    def load_project(self, filename: str, password: str = None, job_ttl: int = None) -> KepServiceResponse:
329        '''Executes a ProjectLoad Service call to the Kepware instance. This loads 
330        a project file to disk.
331
332        INPUTS:
333        
334        :param filename: Fully qualified or relative path and name of project file including the file extension. Absolute
335        paths required for TKS and KEP while file path is relative for TKE:
336
337                Windows - filename = C:\\filepath\\test.opf
338                Linux - filename = filepath/test.lpf - Location is /opt/tkedge/v1/user_data/filepath/test.lpf
339
340        :param password: *(optional)* Specify a password with which to load an encrypted project file.          
341        :param job_ttl: *(optional)* Determines the number of seconds a job instance will exist following completion.
342
343        :return: `KepServiceResponse` instance with job information
344
345        :raises KepHTTPError: If urllib provides an HTTPError (If not HTTP code 202 [Accepted] or 429 [Too Busy] returned)
346        :raises KepURLError: If urllib provides an URLError
347        '''
348        url = self.url + self.__project_services_url + '/ProjectLoad'
349        prop_data = {'servermain.PROJECT_FILENAME': filename}
350        if password != None: prop_data['servermain.PROJECT_PASSWORD'] = password
351        try:
352            job = self._kep_service_execute(url, prop_data, job_ttl)
353            return job
354        except Exception as err:
355            raise err
356
357    def get_project_backup_info(self) -> dict:
358        ''' Get the Project Backup Information of the Kepware instance.
359        
360        :return: List of all the backup projects and their properties.
361
362        :raises KepHTTPError: If urllib provides an HTTPError
363        :raises KepURLError: If urllib provides an URLError
364        '''
365
366        r = self._config_get(self.url + '/project/backups')
367        return r.payload
368
369    def backup_project(self, job_ttl: int = None) -> KepServiceResponse:
370        '''Executes a CreateBackup Service call to the Kepware instance. This saves 
371        a copy of the current project file to disk as a backup that can be retrieved.
372
373        :param job_ttl: *(optional)* Determines the number of seconds a job instance will exist following completion.
374
375        :return: `KepServiceResponse` instance with job information
376
377        :raises KepHTTPError: If urllib provides an HTTPError (If not HTTP code 202 [Accepted] or 429 [Too Busy] returned)
378        :raises KepURLError: If urllib provides an URLError
379        '''
380        url = self.url + self.__project_services_url + '/CreateBackup'
381        try:
382            job = self._kep_service_execute(url, TTL= job_ttl)
383            return job
384        except Exception as err:
385            raise err
386        
387    def service_status(self, resp: KepServiceResponse):
388        '''Returns the status of a service job. Used to verify if a service call
389        has completed or not.
390
391        :param resp: `KepServiceResponse` instance with job information
392
393        :return: `KepServiceStatus` instance with job status
394
395        :raises KepHTTPError: If urllib provides an HTTPError
396        :raises KepURLError: If urllib provides an URLError
397        '''
398        # need to remove part of job href
399        loc = resp.href.find(self.__root_url + self.__version_url)
400        job_url = resp.href[loc + len(self.__root_url + self.__version_url):]
401
402        r = self._config_get(self.url + job_url)
403        job = KepServiceStatus(r.payload['servermain.JOB_COMPLETE'],r.payload['servermain.JOB_STATUS'], r.payload['servermain.JOB_STATUS_MSG'])
404        return job
405
406
407    #Function used to Add an object to Kepware (HTTP POST)
408    def _config_add(self, url, DATA):
409        '''Conducts an POST method at *url* to add an object in the Kepware Configuration
410        *DATA* is required to be a properly JSON object (dict) of the item to be posted to *url* 
411        '''
412        if len(DATA) == 0:
413            err_msg = f'Error: Empty List or Dict in DATA | DATA type: {type(DATA)}'
414            raise KepError(err_msg) 
415        data = json.dumps(DATA).encode('utf-8')
416        url_obj = self.__url_validate(url)
417        q = request.Request(url_obj, data, method='POST')
418        r = self.__connect(q)
419        return r
420
421    #Function used to del an object to Kepware (HTTP DELETE)
422    def _config_del(self, url):
423        '''Conducts an DELETE method at *url* to delete an object in the Kepware Configuration'''
424        url_obj = self.__url_validate(url)
425        q = request.Request(url_obj, method='DELETE')
426        r = self.__connect(q)
427        return r
428
429    #Function used to Update an object to Kepware (HTTP PUT)
430    def _config_update(self, url, DATA = None):
431        '''Conducts an PUT method at *url* to modify an object in the Kepware Configuration.
432        *DATA* is required to be a properly JSON object (dict) of the item to be put to *url*
433        '''
434        url_obj = self.__url_validate(url)
435        if DATA == None:            
436            q = request.Request(url_obj, method='PUT')
437        else:
438            data = json.dumps(DATA).encode('utf-8')
439            q = request.Request(url_obj, data, method='PUT')
440        r = self.__connect(q)
441        return r
442
443    #Function used to Read an object from Kepware (HTTP GET) and return the JSON response
444    def _config_get(self, url, *, params = None):
445        '''
446        Conducts an GET method at *url* to retrieve an objects properties with query parameters in 
447        the Kepware Configuration.
448        '''
449        # Add parameters when necessary
450        if params is not None and params != {}:
451            qparams = parse.urlencode(params)
452            url = f'{url}?{qparams}'
453        url_obj = self.__url_validate(url)
454        q = request.Request(url_obj, method='GET')
455        r = self.__connect(q)
456        return r
457
458    
459    def _force_update_check(self, force, DATA):
460        '''
461        This will validate if the modify call needs to be forced or not. If forced, the dict DATA needs
462        to have the 'FORCE_UPDATE' property with a value of True. If forced is not requested, it is necessary
463        to provide the current 'PROJECT_ID'. If 'PROJECT_ID' is not present in DATA, this will automatically 
464        retreive it from the active server.
465        '''
466        if force == True:
467            DATA['FORCE_UPDATE'] = True
468        else:
469            # Get Project ID if it doesn't exist and if FORCE_UPDATE is existing and FALSE
470            if 'PROJECT_ID' not in DATA:
471                if 'FORCE_UPDATE' in DATA:
472                    if 'FORCE_UPDATE' == False:
473                        try:
474                            project_data = self._config_get(self.url + '/project')
475                            DATA['PROJECT_ID'] = project_data.payload['PROJECT_ID']
476                        except:
477                            #NEED TO COVER ERROR CONDITION
478                            pass
479                else:
480                    try:
481                        project_data = self._config_get(self.url + '/project')
482                        DATA['PROJECT_ID'] = project_data.payload['PROJECT_ID']
483                    except:
484                        #NEED TO COVER ERROR CONDITION
485                        pass
486        return DATA
487    # General service call handler
488    def _kep_service_execute(self, url, DATA = None, TTL = None):
489        try:
490            if TTL != None:
491                if DATA == None: DATA = {}
492                DATA["servermain.JOB_TIME_TO_LIVE_SECONDS"]= TTL
493            r = self._config_update(url, DATA)
494            job = KepServiceResponse(r.payload['code'],r.payload['message'], r.payload['href'])
495            return job
496        except KepHTTPError as err:
497            if err.code == 429:
498                job = KepServiceResponse()
499                job.code = err.code
500                job.message = err.payload
501                return job
502            else:
503                raise err
504
505# 
506# Supporting Functions
507#
508
509    # General connect call to manage HTTP responses for all methods
510    # Returns the response object for the method to handle as appropriate
511    # Raises Errors as found
512    def __connect(self,request_obj):
513        # Fill appropriate header information
514        data = _HttpDataAbstract()
515        request_obj.add_header("Authorization", "Basic %s" % self.__build_auth_str(self.username, self.password))
516        request_obj.add_header("Content-Type", "application/json")
517        request_obj.add_header("Accept", "application/json")
518        try:
519            # context is sent regardless of HTTP or HTTPS - seems to be ignored if HTTP URL
520            with request.urlopen(request_obj, context=self.__ssl_context) as server:
521                try:
522                    payload = server.read()
523                    data.payload = json.loads(codecs.decode(payload,'utf-8-sig'))
524                except:
525                    pass
526                data.code = server.code
527                data.reason = server.reason
528                return data
529        except error.HTTPError as err:
530            payload = json.loads(codecs.decode(err.read(),'utf-8-sig'))
531            # print('HTTP Code: {}\n{}'.format(err.code,payload), file=sys.stderr)
532            raise KepHTTPError(url=err.url, code=err.code, msg=err.msg, hdrs=err.hdrs, payload=payload)
533        except error.URLError as err:
534            # print('URLError: {} URL: {}'.format(err.reason, request_obj.get_full_url()), file=sys.stderr)
535            raise KepURLError(msg=err.reason, url=request_obj.get_full_url())
536
537    # Fucntion used to ensure special characters are handled in the URL
538    # Ex: Space will be turned to %20
539    def __url_validate(self, url):
540        # Configuration API does not use fragments in URL so ignore to allow # as a character
541        # Objects in Kepware can include # as part of the object names
542        parsed_url = parse.urlparse(url, allow_fragments= False)
543        # Added % for scenarios where special characters have already been escaped with %
544        updated_path = parse.quote(parsed_url.path, safe = '/%')
545
546        # If host is "localhost", force using the IPv4 loopback adapter IP address in all calls
547        # This is done to remove retries that will happen when the host resolution uses IPv6 intially
548        # Kepware currently doesn't support IPv6 and is not listening on this interface
549        if parsed_url.hostname.lower() == 'localhost':
550            ip = socket.gethostbyname(parsed_url.hostname)
551            parsed_url = parsed_url._replace(netloc='{}:{}'.format(ip, parsed_url.port))
552        
553        return parsed_url._replace(path=updated_path).geturl()
554
555    # Function used to build the basic authentication string
556    def __build_auth_str(self, username, password):
557        if isinstance(username, str):
558            username = username.encode('latin1')
559        if isinstance(password, str):
560            password = password.encode('latin1')
561        authstr = b64encode(b':'.join((username, password))).strip().decode('ascii')
562        return authstr
563    
564    # Create parameters for log queries
565    def __create_query(self, start = None, end = None, limit = None):
566        query = {}
567        if start != None and isinstance(start, datetime.datetime):
568            query['start'] = start.isoformat()
569        if end != None and isinstance(end, datetime.datetime):
570            query['end'] = end.isoformat()
571        if limit != None:
572            query['limit'] = limit
573        return query
574
575    # Create filter query for log queries
576    def __create_filter_query(self, filters: list[Filter] = None):
577        query = {}
578        if filters is None:
579            return query
580        for f in filters:
581            # Ensure we use the value of the Enum, not the Enum object itself
582            key = f"filter[{f.field.value}][{f.modifier.value}]"
583            query[key] = f.value
584        return query