Added methods for instance certificate management changes in UAG

Author: rlabbeptc

kepconfig/ua_gateway/client.py

@@ -11,9 +11,13 @@
 """
 
 from typing import Union
+
+from kepconfig.structures import KepServiceResponse
 from ..connection import server
 from ..error import KepError, KepHTTPError
-from ..ua_gateway.common import _INTER_TYPE, _change_cert_trust,  _create_url_cert, _create_url_client, _delete_cert_truststore
+from ..ua_gateway.common import _INTER_TYPE, _change_cert_trust,  _create_url_cert, _create_url_client, _delete_cert_truststore, _create_url_inst_cert
+
+CLIENT_INSTANCE_CERTIFICATE = "Client Instance Certificate"
 
 def get_ua_client_connection(server: server, ua_client_connection: str) -> dict:
     '''Returns the properties of the UAG client connection object.
@@ -192,4 +196,37 @@ def delete_certificate(server: server, certificate: str) -> bool:
     :raises KepHTTPError: If urllib provides an HTTPError
     :raises KepURLError: If urllib provides an URLError
     '''
-    return _delete_cert_truststore(server, _INTER_TYPE.CLIENT, certificate)
+    return _delete_cert_truststore(server, _INTER_TYPE.CLIENT, certificate)
+
+def get_instance_certificate(server: server) -> dict:
+    '''Returns the properties of the UAG client instance certificate object in the UAG certificate store. 
+    These are UAG instance certificates that are used by UAG for trust purposes in the UA security model.
+    
+    :param server: instance of the `server` class
+    
+    :return: Dict of properties for the certificate requested
+
+    :raises KepHTTPError: If urllib provides an HTTPError
+    :raises KepURLError: If urllib provides an URLError
+    '''
+    r = server._config_get(server.url + _create_url_inst_cert(_INTER_TYPE.CLIENT, CLIENT_INSTANCE_CERTIFICATE))
+    return r.payload
+
+def reissue_self_signed_instance_certificate(server: server, job_ttl: int = None) -> KepServiceResponse:
+    '''Deletes and reissues a self-signed UAG server instance certificate object in the UAG certificate store. 
+    This is the UAG instance certificate that are used by UAG for trust purposes in the UA security model.
+    
+    :param server: instance of the `server` class
+    :param job_ttl: *(optional)* Determines the number of seconds a job instance will exist following completion.
+
+    :return: `KepServiceResponse` instance with job information
+
+    :raises KepHTTPError: If urllib provides an HTTPError
+    :raises KepURLError: If urllib provides an URLError
+    '''
+    url = server.url + _create_url_inst_cert(_INTER_TYPE.CLIENT, CLIENT_INSTANCE_CERTIFICATE) + '/services/ReIssueInstanceCertificate'
+    try:
+        job = server._kep_service_execute(url, TTL= job_ttl)
+        return job
+    except Exception as err:
+        raise err

kepconfig/ua_gateway/common.py

@@ -16,23 +16,24 @@
 CLIENT_ROOT = f'{UA_GATEWAY_ROOT}/ua_client_interfaces/Client Interface'
 CONN_ROOT = f'{CLIENT_ROOT}/ua_client_connections'
 CLIENT_CERT_ROOT = f'{CLIENT_ROOT}/certificates'
+CLIENT_INST_CERT_ROOT = f'{CLIENT_ROOT}/client_instance_certificates'
 SERVER_ROOT = f'{UA_GATEWAY_ROOT}/ua_server_interfaces/Server Interface'
 ENDPOINT_ROOT = f'{SERVER_ROOT}/ua_server_endpoints'
 SERVER_CERT_ROOT = f'{SERVER_ROOT}/certificates'
+SERVER_INST_CERT_ROOT = f'{SERVER_ROOT}/server_instance_certificates'
 
 
 class _INTER_TYPE(Enum):
     SERVER = 0
     CLIENT = 1
     CERTS = 2
 
+# TODO: DEPRECATED: This constant is deprecated and will be removed in a future release.
 INSTANCE_CERTIFICATE = "Instance Certificate"
 
 def _create_url_cert(interface, certificate = None):
     '''Creates url object for the "certificate" branch of Kepware's UA Gateway. Used 
     to build a part of Kepware Configuration API URL structure
-
-    Returns the UA Gateway client interfaces specific certificate url when a value is passed as the certificate name.
     '''
     if interface == _INTER_TYPE.SERVER:
         if certificate == None:
@@ -44,6 +45,27 @@ def _create_url_cert(interface, certificate = None):
             return CLIENT_CERT_ROOT
         else:
             return f'{CLIENT_CERT_ROOT}/{_url_parse_object(certificate)}'
+    # TODO: DEPRECATED: This interface type is deprecated and will be removed in a future release.
+    else:
+        if certificate == None:
+            return CERT_ROOT
+        else:
+            return '{}/{}'.format(CERT_ROOT,_url_parse_object(certificate))
+
+def _create_url_inst_cert(interface, certificate = None):
+    '''Creates url object for the "instance certificate" branch of Kepware's UA Gateway interfaces. Used 
+    to build a part of Kepware Configuration API URL structure
+    '''
+    if interface == _INTER_TYPE.SERVER:
+        if certificate == None:
+            return SERVER_INST_CERT_ROOT
+        else:
+            return f'{SERVER_INST_CERT_ROOT}/{_url_parse_object(certificate)}'
+    elif interface == _INTER_TYPE.CLIENT:
+        if certificate == None:
+            return CLIENT_INST_CERT_ROOT
+        else:
+            return f'{CLIENT_INST_CERT_ROOT}/{_url_parse_object(certificate)}'
     else:
         if certificate == None:
             return CERT_ROOT

kepconfig/ua_gateway/server.py

@@ -11,9 +11,13 @@
 """
 
 from typing import Union
+
+from kepconfig.structures import KepServiceResponse
 from ..connection import server
 from ..error import KepError, KepHTTPError
-from ..ua_gateway.common import _INTER_TYPE, _change_cert_trust, _create_url_cert, _create_url_server, _delete_cert_truststore, SERVER_ROOT
+from ..ua_gateway.common import _INTER_TYPE, _change_cert_trust, _create_url_cert, _create_url_server, _delete_cert_truststore, SERVER_ROOT, _create_url_inst_cert
+
+SERVER_INSTANCE_CERTIFICATE = 'Server Instance Certificate'
 
 def get_uag_server_interface_properties(server: server) -> dict:
     ''' Get the UAG Server Interface Properties of the Kepware instance. These properties expose User Identify
@@ -224,4 +228,37 @@ def delete_certificate(server: server, certificate: str) -> bool:
     :raises KepHTTPError: If urllib provides an HTTPError
     :raises KepURLError: If urllib provides an URLError
     '''
-    return _delete_cert_truststore(server, _INTER_TYPE.SERVER, certificate)
+    return _delete_cert_truststore(server, _INTER_TYPE.SERVER, certificate)
+
+def get_instance_certificate(server: server) -> dict:
+    '''Returns the properties of the UAG server instance certificate object in the UAG certificate store. 
+    These are UAG instance certificates that are used by UAG for trust purposes in the UA security model.
+    
+    :param server: instance of the `server` class
+    
+    :return: Dict of properties for the certificate requested
+
+    :raises KepHTTPError: If urllib provides an HTTPError
+    :raises KepURLError: If urllib provides an URLError
+    '''
+    r = server._config_get(server.url + _create_url_inst_cert(_INTER_TYPE.SERVER, SERVER_INSTANCE_CERTIFICATE))
+    return r.payload
+
+def reissue_self_signed_instance_certificate(server: server, job_ttl: int = None) -> KepServiceResponse:
+    '''Deletes and reissues a self-signed UAG server instance certificate object in the UAG certificate store. 
+    This is the UAG instance certificate that are used by UAG for trust purposes in the UA security model.
+    
+    :param server: instance of the `server` class
+    :param job_ttl: *(optional)* Determines the number of seconds a job instance will exist following completion.
+
+    :return: `KepServiceResponse` instance with job information
+
+    :raises KepHTTPError: If urllib provides an HTTPError
+    :raises KepURLError: If urllib provides an URLError
+    '''
+    url = server.url + _create_url_inst_cert(_INTER_TYPE.SERVER, SERVER_INSTANCE_CERTIFICATE) + '/services/ReIssueInstanceCertificate'
+    try:
+        job = server._kep_service_execute(url, TTL= job_ttl)
+        return job
+    except Exception as err:
+        raise err