feat: add authorization with token

indykite_sdk/api.py

@@ -11,6 +11,7 @@
 from indykite_sdk.utils.hash_methods import encrypt_bcrypt, encrypt_sha256
 from indykite_sdk.identity import IdentityClient
 from indykite_sdk.config import ConfigClient
+from indykite_sdk.authorization import AuthorizationClient
 from indykite_sdk.indykite.config.v1beta1.model_pb2 import (SendGridProviderConfig, MailJetProviderConfig, AmazonSESProviderConfig, MailgunProviderConfig)
 from indykite_sdk.indykite.config.v1beta1.model_pb2 import (EmailServiceConfig, AuthFlowConfig, OAuth2ClientConfig, IngestMappingConfig)
 from indykite_sdk.indykite.config.v1beta1.model_pb2 import OAuth2ProviderConfig, OAuth2ApplicationConfig
@@ -19,6 +20,7 @@
 from indykite_sdk.indykite.config.v1beta1.model_pb2 import EmailAttachment, Email, EmailMessage, EmailTemplate, EmailDefinition
 from indykite_sdk.indykite.config.v1beta1.model_pb2 import google_dot_protobuf_dot_wrappers__pb2 as wrappers
 from indykite_sdk.indykite.identity.v1beta2.import_pb2 import Email as EmailIdentity
+from indykite_sdk.model.is_authorized import IsAuthorizedResource
 
 
 class ParseKwargs(argparse.Action):
@@ -455,10 +457,25 @@ def main():
     import_digital_twins_update_parser.add_argument("id", help="Digital Twin id (gid)")
     import_digital_twins_update_parser.add_argument("tenant_id", help="Tenant id (gid)")
 
+    # is_authorized_dt
+    is_authorized_dt_parser = subparsers.add_parser("is_authorized_dt")
+    is_authorized_dt_parser.add_argument("digital_twin_id", help="Digital Twin id (gid)")
+    is_authorized_dt_parser.add_argument("tenant_id", help="Tenant id (gid)")
+
+    # is_authorized_token
+    is_authorized_token_parser = subparsers.add_parser("is_authorized_token")
+    is_authorized_token_parser.add_argument("access_token")
+
+    # is_authorized_property
+    is_authorized_property_parser = subparsers.add_parser("is_authorized_property")
+    is_authorized_property_parser.add_argument("email", help="Existing Digital Twin email")
+    is_authorized_property_parser.add_argument("tenant_id", help="Tenant id (gid)")
+
     args = parser.parse_args()
     local = args.local
     client = IdentityClient(local)
     client_config = ConfigClient(local)
+    client_authorization = AuthorizationClient(local)
 
     command = args.command
 
@@ -1624,6 +1641,17 @@ def main():
             print("Invalid import digital twins response")
         return import_digital_twins_config_response
 
+    elif command == "is_authorized_token":
+        access_token = args.access_token
+        actions = ["HAS_FREE_PARKING"]
+        resources = [IsAuthorizedResource("lotA", "ParkingLot"), IsAuthorizedResource("lotB", "ParkingLot")]
+        is_authorized = client_authorization.is_authorized_token(access_token, resources, actions)
+        if is_authorized:
+            print_response(is_authorized)
+        else:
+            print("Invalid is_authorized")
+        return is_authorized
+
 
 def print_verify_info(digital_twin_info):  # pragma: no cover
     print("Digital twin info")

indykite_sdk/authorization/__init__.py

@@ -0,0 +1,52 @@
+import certifi
+import grpc
+import os
+
+from indykite_sdk.authorization import helper
+from indykite_sdk.indykite.identity.v1beta1 import identity_management_api_pb2_grpc as pb2_grpc
+
+
+class AuthorizationClient(object):
+
+    def __init__(self, local=False):
+        cred = os.getenv('INDYKITE_APPLICATION_CREDENTIALS')
+
+        # Load the config from File (secondary)
+        if (cred is False) or (cred is None):
+            cred = os.getenv('INDYKITE_APPLICATION_CREDENTIALS_FILE')
+            try:
+                if (cred is False) or (cred is None):
+                    raise Exception("Missing INDYKITE_APPLICATION_CREDENTIALS or "
+                                    "INDYKITE_APPLICATION_CREDENTIALS_FILE environment variable")
+            except Exception as exception:
+                print(exception)
+                return None
+            credentials = os.path.join(os.path.dirname(cred), os.path.basename(cred))
+            credentials = helper.load_credentials(credentials)
+
+        # Load the credential json (primary)
+        else:
+            credentials = helper.load_json(cred)
+
+        agent_token = helper.create_agent_jwt(credentials)
+
+        call_credentials = grpc.access_token_call_credentials(agent_token.decode("utf-8"))
+
+        if local:
+            certificate_path = os.getenv('CAPEM')
+            endpoint = credentials.get("local_endpoint")
+        else:
+            certificate_path = certifi.where()
+            endpoint = credentials.get("endpoint")
+
+        with open(certificate_path, "rb") as cert_file:
+            channel_credentials = grpc.ssl_channel_credentials(cert_file.read())
+
+        composite_credentials = grpc.composite_channel_credentials(channel_credentials,
+                                                                   call_credentials)
+
+        self.channel = grpc.secure_channel(endpoint, composite_credentials)
+        self.stub = pb2_grpc.IdentityManagementAPIStub(channel=self.channel)
+
+    # Imported methods
+    from .is_authorized import is_authorized_token

indykite_sdk/authorization/helper.py

@@ -0,0 +1,43 @@
+import json
+import time
+import uuid
+from authlib.jose import JsonWebKey, jwt
+from datetime import datetime, timedelta, timezone
+
+
+def load_credentials(path):
+    with open(path, 'r') as file:
+        raw_content = file.read()
+    return json.loads(raw_content)
+
+
+def load_json(content):
+    return json.loads(content)
+
+
+def create_agent_jwt(credentials):
+    jwk = credentials.get('privateKeyJWK')
+    key = JsonWebKey.import_key(jwk)
+    message = create_jwt_message(credentials)
+    jwt_token = jwt.encode({
+            'alg': 'ES256',
+            'cty': 'JWT',
+            'kid': jwk['kid']
+    }, message, key)
+    return jwt_token
+
+
+def create_jwt_message(credentials):
+    message = {
+            'exp': get_int_from_datetime(datetime.now(timezone.utc) + timedelta(hours=24)),
+            'iat': get_int_from_datetime(datetime.now(timezone.utc)),
+            'iss': credentials.get('appAgentId'),
+            'jti': str(uuid.uuid4()),
+            'sub': credentials.get('appAgentId'),
+    }
+    return message
+
+
+def get_int_from_datetime(dt):
+    return int(time.mktime(dt.timetuple()))
+

indykite_sdk/authorization/is_authorized.py

@@ -0,0 +1,30 @@
+from indykite_sdk.indykite.authorization.v1beta1 import authorization_service_pb2 as pb2
+from indykite_sdk.indykite.identity.v1beta2 import identity_management_api_pb2 as pb2_ident
+
+
+def is_authorized_token(self, access_token, resources=[], actions=[]):
+    try:
+        response = self.stub.IsAuthorized(
+            pb2.IsAuthorizedRequest(
+                digital_twin_identifier=pb2_ident.DigitalTwinIdentifier(
+                    access_token=str(access_token)
+                ),
+                resources=request_resource(resources),
+                actions=actions
+            )
+        )
+    except Exception as exception:
+        print(exception)
+        return None
+
+    if not response:
+        return None
+
+    return response
+
+
+def request_resource(resources):
+    res = []
+    for r in resources:
+        res.append(pb2.IsAuthorizedRequest.Resource(id=r.id, label=r.label))
+    return res

indykite_sdk/model/is_authorized.py

@@ -0,0 +1,36 @@
+from indykite_sdk.utils import timestamp_to_date
+from google.protobuf.json_format import MessageToDict
+
+
+class IsAuthorizedResponse:
+    @classmethod
+    def deserialize(cls, message):
+        if message is None:
+            return None
+
+        is_authorized_response = IsAuthorizedResponse(
+            decision_time=timestamp_to_date(message.decision_time),
+            decisions=MessageToDict(message.decisions)
+        )
+
+        return is_authorized_response
+
+    def __init__(self, decision_time, decisions):
+        self.decision_time = decision_time,
+        self.decisions = decisions
+
+
+class IsAuthorizedResource:
+
+    def __init__(self, id: any, label):
+        self.id = id
+        self.label = label
+
+
+class IsAuthorizedDecisions:
+
+    def __init__(self, id: any, label):
+        self.id = id
+        self.label = label
+
+

tests/test_import_digital_twins.py

@@ -565,6 +565,7 @@ class Entity:
     response = client.import_digital_twins(entities, hash_algorithm)
     assert response is None
 
+
 def test_import_digital_twin_non_valid_user_info2():
     right_now = str(int(time.time()))
     tenant_id = data.get_tenant()

tests/test_is_authorized.py

@@ -0,0 +1,44 @@
+import json
+import time
+from datetime import datetime
+
+from indykite_sdk.authorization import AuthorizationClient
+from indykite_sdk.indykite.authorization.v1beta1 import authorization_service_pb2 as pb2
+from indykite_sdk.indykite.identity.v1beta2 import identity_management_api_pb2 as pb2_ident
+from google.protobuf.json_format import MessageToDict
+from indykite_sdk.model.is_authorized import IsAuthorizedResource
+from helpers import data
+
+
+def test_is_authorized_token_wrong_token():
+    client = AuthorizationClient()
+    assert client is not None
+
+    access_token = data.get_expired_token()
+    actions = ["HAS_FREE_PARKING"]
+    resources = [IsAuthorizedResource("lotA", "ParkingLot"), IsAuthorizedResource("lotB", "ParkingLot")]
+    response = client.is_authorized_token(access_token, resources, actions)
+    assert response is None
+
+
+def test_is_authorized_token_success():
+    client = AuthorizationClient()
+    assert client is not None
+
+    access_token = data.get_verification_bearer()
+    actions = ["HAS_FREE_PARKING"]
+    resources = [IsAuthorizedResource("lotA", "ParkingLot"), IsAuthorizedResource("lotB", "ParkingLot")]
+    res = []
+    for r in resources:
+        res.append(pb2.IsAuthorizedRequest.Resource(id=r.id, label=r.label))
+    digital_twin_identifier = pb2_ident.DigitalTwinIdentifier(
+        access_token=str(access_token)
+    )
+
+    def mocked_is_authorized(request: pb2.IsAuthorizedRequest):
+        assert request.digital_twin_identifier == digital_twin_identifier
+        return pb2.IsAuthorizedResponse()
+
+    client.stub.IsAuthorized = mocked_is_authorized
+    response = client.is_authorized_token(access_token, resources, actions)
+    assert response is not None