feat: custom login token

.coveragerc

@@ -19,6 +19,7 @@ omit =
   indykite_sdk/utils/logger.py
   tests/*
   examples/*
+  prepare_docs/*
 
 [report]
 exclude_lines =

codecov.yaml

@@ -25,3 +25,4 @@ ignore:
   - "indykite_sdk/utils/logger.py"
   - "tests"
   - "examples"
+  - "prepare_docs"

indykite_sdk/api.py

@@ -20,7 +20,7 @@
                                                             ReadIDProviderConfig, KnowledgeGraphSchemaConfig )
 from indykite_sdk.indykite.config.v1beta1.model_pb2 import OAuth2ProviderConfig, OAuth2ApplicationConfig
 from indykite_sdk.indykite.identity.v1beta2.import_pb2 import ImportDigitalTwinsRequest, ImportDigitalTwin, \
-    ImportProperties, UserProvider, UserMetadata
+    ImportProperties, UserProvider, UserMetadata, CredentialReference
 from indykite_sdk.indykite.identity.v1beta2.import_pb2 import PasswordCredential, PasswordHash, Bcrypt, SHA256
 from indykite_sdk.indykite.config.v1beta1.model_pb2 import EmailAttachment, Email, EmailMessage, EmailTemplate, \
     EmailDefinition
@@ -32,6 +32,7 @@
 from indykite_sdk.model.tenant import Tenant
 from indykite_sdk.indykite.identity.v1beta2 import attributes_pb2 as attributes
 from indykite_sdk.ingest import IngestClient
+from indykite_sdk.indykite.identity.v1beta2 import model_pb2 as model
 from indykite_sdk.identity import helper
 import logging
 from indykite_sdk.utils.message_to_value import arg_to_value
@@ -661,6 +662,17 @@ def main():
     # get_schema_helpers
     get_schema_helpers_parser = subparsers.add_parser("get_schema_helpers")
 
+    # create-custom-login-token
+    create_custom_login_token = subparsers.add_parser("create-custom-login-token")
+    create_custom_login_token.add_argument("dt_id", help="DigitalTwin gid id")
+    create_custom_login_token.add_argument("tenant_id", help="Tenant gid id")
+
+    # create-custom-login-token-property
+    create_custom_login_token_property = subparsers.add_parser("create-custom-login-token-property")
+    create_custom_login_token_property.add_argument("type", help="property_filter type")
+    create_custom_login_token_property.add_argument("value", help="property_filter value")
+    create_custom_login_token_property.add_argument("tenant_id", help="Tenant gid id")
+
     args = parser.parse_args()
     local = args.local
     client = IdentityClient(local)
@@ -856,20 +868,13 @@ def main():
             state="DIGITAL_TWIN_STATE_ACTIVE",
             password=PasswordCredential(
                 email=EmailIdentity(
-                    email="test2214@example.com",
+                    email="test2314@example.com",
                     verified=True
                 ),
                 value="password"
             ),
             tags=[],
-            provider_user_info=[
-                UserProvider(
-                    uid="23456789456",
-                    provider_id="gid:AAAAEezCvUQGV0HgotmCoeCJAck",
-                    email="test@example.com",
-                    display_name="Provider name",
-                    photo_url="https://url_photo"
-                )],
+            provider_user_info=[],
             properties=ImportProperties(
                 operations=[attributes.PropertyBatchOperation(
                     add=attributes.Property(
@@ -890,7 +895,7 @@ def main():
                 state="DIGITAL_TWIN_STATE_ACTIVE",
                 password=PasswordCredential(
                     email=EmailIdentity(
-                        email="test2215@example.com",
+                        email="test2315@example.com",
                         verified=True
                     ),
                     value="password"
@@ -902,7 +907,7 @@ def main():
                 state="DIGITAL_TWIN_STATE_ACTIVE",
                 password=PasswordCredential(
                     email=EmailIdentity(
-                        email="test2216@example.com",
+                        email="test2316@example.com",
                         verified=True
                     ),
                     value="password"
@@ -2734,6 +2739,56 @@ def main():
             print("Invalid get schema helpers")
         return get_schema_helpers
 
+    elif command == "create-custom-login-token":
+        digital_twin_id = args.dt_id
+        tenant_id = args.tenant_id
+        token_claims = {"t_claim": "test"}
+        session_claims = {"s_claim": "test"}
+        digital_twin = model.DigitalTwin(
+                    id=str(digital_twin_id),
+                    tenant_id=str(tenant_id)
+                )
+        create_custom_login_token = client.create_custom_login_token(digital_twin, token_claims, session_claims)
+        if create_custom_login_token:
+            api_helper.print_response(create_custom_login_token)
+        else:
+            print("Invalid custom login")
+        return create_custom_login_token
+
+    elif command == "create-custom-login-token-property":
+        type = args.type
+        value = args.value
+        tenant_id = args.tenant_id
+        property_filter = client.property_filter(type, value, tenant_id)
+        token_claims = {"t_claim": "test"}
+        session_claims = {"s_claim": "test"}
+
+        create_custom_login_token = client.create_custom_login_token(property_filter, token_claims, session_claims)
+        if create_custom_login_token:
+            api_helper.print_response(create_custom_login_token)
+        else:
+            print("Invalid custom login")
+        return create_custom_login_token
+
+    elif command == "create-custom-login-token-credential":
+        # ProviderId identifies the credential provider which the uid belongs to (password, webauthn, google.com...)
+        provider_id = args.provider_id
+        # Uid is the unique identifier of subject in the external identity provider referenced by ProviderId
+        uid = args.uid
+        credential_reference = CredentialReference(
+            provider_id=str(provider_id),
+            uid=str(uid)
+        )
+        token_claims = {"t_claim": "test"}
+        session_claims = {"s_claim": "test"}
+
+        create_custom_login_token = client.create_custom_login_token(credential_reference, token_claims, session_claims)
+        if create_custom_login_token:
+            api_helper.print_response(create_custom_login_token)
+        else:
+            print("Invalid custom login")
+        return create_custom_login_token
+
 
 if __name__ == '__main__':  # pragma: no cover
     main()

indykite_sdk/identity/__init__.py

@@ -64,3 +64,4 @@ def __init__(self, local=False):
     from .forgotten_password import start_forgotten_password_flow
     from .invitation import create_email_invitation, create_mobile_invitation, check_invitation_state, resend_invitation, cancel_invitation
     from .register_digital_twins_no_cred import register_digital_twin_without_credential
+    from .create_custom_login_token import create_custom_login_token

indykite_sdk/identity/create_custom_login_token.py

@@ -0,0 +1,62 @@
+from indykite_sdk.indykite.identity.v1beta2 import identity_management_api_pb2 as pb2
+from google.protobuf import struct_pb2
+import sys
+import indykite_sdk.utils.logger as logger
+from indykite_sdk.indykite.identity.v1beta2 import attributes_pb2 as attributes
+from indykite_sdk.indykite.identity.v1beta2 import model_pb2 as model
+from indykite_sdk.indykite.identity.v1beta2 import import_pb2
+from indykite_sdk.model.create_custom_login_token import CreateCustomLoginToken
+
+
+def create_custom_login_token(self, uid: any, token_claims: dict, session_claims: dict):
+    """
+    CreateCustomLoginToken creates a signed custom authentication token with the specified user ID
+    :param self:
+    :param uid: DigitalTwin, PropertyFilter or CredentialReference
+    :param token_claims: dictionary
+    :param session_claims: dict
+    :return: deserialized CreateCustomLoginTokenResponse
+    """
+    sys.excepthook = logger.handle_excepthook
+    try:
+        token_struct = struct_pb2.Struct()
+        if token_claims is not None:
+            token_struct.update(token_claims)
+
+        session_struct = struct_pb2.Struct()
+        if session_claims is not None:
+            session_struct.update(session_claims)
+
+        if isinstance(uid, model.DigitalTwin):
+            response = self.stub.CreateCustomLoginToken(
+                pb2.CreateCustomLoginTokenRequest(
+                    digital_twin=uid,
+                    token_claims=token_struct,
+                    session_claims=session_struct,
+                )
+            )
+        elif isinstance(uid, attributes.PropertyFilter):
+            response = self.stub.CreateCustomLoginToken(
+                pb2.CreateCustomLoginTokenRequest(
+                    property_filter=uid,
+                    token_claims=token_struct,
+                    session_claims=session_struct,
+                )
+            )
+        elif isinstance(uid, import_pb2.CredentialReference):
+            response = self.stub.CreateCustomLoginToken(
+                pb2.CreateCustomLoginTokenRequest(
+                    credential_reference=uid,
+                    token_claims=token_struct,
+                    session_claims=session_struct,
+                )
+            )
+        else:
+            raise Exception("uid is not valid")
+        return CreateCustomLoginToken.deserialize(response) if response else None
+
+    except Exception as exception:
+        return logger.logger_error(exception)
+
+    if not response:
+        return None

indykite_sdk/model/create_custom_login_token.py

@@ -0,0 +1,17 @@
+from indykite_sdk.model.digital_twin import DigitalTwinCore
+
+
+class CreateCustomLoginToken:
+    @classmethod
+    def deserialize(cls, message):
+        if message is None:
+            return None
+        create_custom_login_token = CreateCustomLoginToken(
+            str(message.token),
+            DigitalTwinCore.deserialize(message.digital_twin)
+        )
+        return create_custom_login_token
+
+    def __init__(self, token, digital_twin):
+        self.token = token
+        self.digital_twin = digital_twin

indykite_sdk/model/digital_twin.py

@@ -12,14 +12,16 @@ def deserialize(cls, message):
             str(message.id),
             str(message.tenant_id),
             message.kind,
-            message.state
+            message.state,
+            message.tags if message.tags else None
         )
 
-    def __init__(self, dt_id, tenant_id, kind, state):
+    def __init__(self, dt_id, tenant_id, kind, state, tags=None):
         self.id = dt_id
         self.tenantId = tenant_id
         self.kind = kind
         self.state = state
+        self.tags = tags
 
     def __str__(self):
         return (

tests/test_create_custom_login_token.py

@@ -0,0 +1,74 @@
+from indykite_sdk.identity import IdentityClient
+from indykite_sdk.indykite.identity.v1beta2 import identity_management_api_pb2 as pb2
+from indykite_sdk.indykite.identity.v1beta2 import model_pb2 as model
+from indykite_sdk.indykite.identity.v1beta2.import_pb2 import CredentialReference
+import os
+
+
+def test_create_custom_login_token_error(capsys):
+    client = IdentityClient()
+    assert client is not None
+    digital_twin = model.DigitalTwin(
+        id=str("ANYTHING"),
+        tenant_id=str(os.getenv('ANYTHING'))
+    )
+
+    def mocked_create_custom_login_token_error(request: pb2.CreateCustomLoginTokenRequest):
+        raise Exception("something went wrong")
+
+    client.stub.CreateCustomLoginToken = mocked_create_custom_login_token_error
+
+    response = client.create_custom_login_token(digital_twin, None, None)
+    captured = capsys.readouterr()
+    assert "something went wrong" in captured.err
+
+
+def test_create_custom_login_token():
+    digital_twin = model.DigitalTwin(
+        id=str(os.getenv('DIGITAL_TWIN')),
+        tenant_id=str(os.getenv('TENANT_ID'))
+    )
+    client = IdentityClient()
+    assert client is not None
+
+    def mocked_create_custom_login_token(request: pb2.CreateCustomLoginTokenRequest):
+        assert request.digital_twin == digital_twin
+        return pb2.CreateCustomLoginTokenResponse()
+
+    client.stub.CreateCustomLoginToken = mocked_create_custom_login_token
+    response = client.create_custom_login_token(digital_twin, {"t_claim": "test"}, {"s_claim": "test"})
+
+    assert response is not None
+
+
+def test_create_custom_login_token_credential():
+    credential_reference = CredentialReference(
+        provider_id=str(os.getenv('OAUTH2_PROVIDER')),
+        uid=str("unique_identifier")
+    )
+    client = IdentityClient()
+    assert client is not None
+
+    def mocked_create_custom_login_token(request: pb2.CreateCustomLoginTokenRequest):
+        assert request.credential_reference == credential_reference
+        return pb2.CreateCustomLoginTokenResponse()
+
+    client.stub.CreateCustomLoginToken = mocked_create_custom_login_token
+    response = client.create_custom_login_token(credential_reference, {"t_claim": "test"}, {"s_claim": "test"})
+
+    assert response is not None
+
+
+def test_create_custom_login_token_property():
+    client = IdentityClient()
+    assert client is not None
+    property_filter = client.property_filter("email", "testem@example.com", os.getenv('TENANT_ID'))
+
+    def mocked_create_custom_login_token(request: pb2.CreateCustomLoginTokenRequest):
+        assert request.property_filter == property_filter
+        return pb2.CreateCustomLoginTokenResponse()
+
+    client.stub.CreateCustomLoginToken = mocked_create_custom_login_token
+    response = client.create_custom_login_token(property_filter, {"t_claim": "test"}, {"s_claim": "test"})
+
+    assert response is not None