18284 feat: digital credentials

legal-api/migrations/versions/6e28f267db2a_create_issued_business_user_credentials_.py

@@ -0,0 +1,30 @@
+"""create issued business user credentials table
+
+Revision ID: 6e28f267db2a
+Revises: 8148a25d695e
+Create Date: 2023-10-17 02:17:08.232290
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '6e28f267db2a'
+down_revision = '8148a25d695e'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table('dc_issued_business_user_credentials',
+    sa.Column('id', sa.Integer(), nullable=False),
+    sa.Column('user_id', sa.Integer(), nullable=False),
+    sa.Column('business_id', sa.Integer(), nullable=False),
+    sa.PrimaryKeyConstraint('id'),
+    sa.ForeignKeyConstraint(['business_id'], ['businesses.id']),
+    sa.ForeignKeyConstraint(['user_id'], ['users.id']))
+
+
+def downgrade():
+    op.drop_table('dc_issued_business_user_credentials')

legal-api/src/legal_api/models/__init__.py

@@ -23,6 +23,7 @@
 from .corp_type import CorpType
 from .dc_connection import DCConnection
 from .dc_definition import DCDefinition
+from .dc_issued_business_user_credential import DCIssuedBusinessUserCredential
 from .dc_issued_credential import DCIssuedCredential
 from .document import Document, DocumentType
 from .filing import Filing
@@ -40,6 +41,7 @@
 
 __all__ = ('db',
            'Address', 'Alias', 'Business', 'ColinLastUpdate', 'Comment', 'ConsentContinuationOut', 'CorpType',
-           'DCConnection', 'DCDefinition', 'DCIssuedCredential', 'Document', 'DocumentType',
-           'Filing', 'Office', 'OfficeType', 'Party', 'RegistrationBootstrap', 'RequestTracker', 'Resolution',
-           'PartyRole', 'ShareClass', 'ShareSeries', 'User', 'UserRoles', 'NaicsStructure', 'NaicsElement')
+           'DCConnection', 'DCDefinition', 'DCIssuedCredential', 'DCIssuedBusinessUserCredential', 'Document',
+           'DocumentType', 'Filing', 'Office', 'OfficeType', 'Party', 'RegistrationBootstrap', 'RequestTracker',
+           'Resolution', 'PartyRole', 'ShareClass', 'ShareSeries', 'User', 'UserRoles', 'NaicsStructure',
+           'NaicsElement')

legal-api/src/legal_api/models/dc_issued_business_user_credential.py

@@ -0,0 +1,49 @@
+# Copyright © 2023 Province of British Columbia
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""This module holds data for issued credential."""
+from __future__ import annotations
+
+from typing import List
+
+from .db import db
+
+
+class DCIssuedBusinessUserCredential(db.Model):  # pylint: disable=too-many-instance-attributes
+    """This class manages the issued credential IDs for a user of a business."""
+
+    __tablename__ = 'dc_issued_business_user_credentials'
+
+    id = db.Column(db.Integer, primary_key=True)
+
+    user_id = db.Column('user_id', db.Integer, db.ForeignKey('users.id'))
+    business_id = db.Column('business_id', db.Integer, db.ForeignKey('businesses.id'))
+
+    def save(self):
+        """Save the object to the database immediately."""
+        db.session.add(self)
+        db.session.commit()
+
+    @classmethod
+    def find_by(cls,
+                business_id: int = None,
+                user_id: int = None) -> List[DCIssuedBusinessUserCredential]:
+        """Return the issued business user credential matching the user_id and buisness_id."""
+        dc_issued_business_user_credential = None
+        if business_id and user_id:
+            dc_issued_business_user_credential = (
+                cls.query
+                .filter(DCIssuedBusinessUserCredential.business_id == business_id)
+                .filter(DCIssuedBusinessUserCredential.user_id == user_id)
+                .one_or_none())
+        return dc_issued_business_user_credential

legal-api/src/legal_api/models/dc_issued_credential.py

@@ -48,7 +48,7 @@ def json(self):
             'credentialExchangeId': self.credential_exchange_id,
             'credentialId': self.credential_id,
             'isIssued': self.is_issued,
-            'dateOfIssue': self.date_of_issue.isoformat(),
+            'dateOfIssue': self.date_of_issue.isoformat() if self.date_of_issue else None,
             'isRevoked': self.is_revoked,
             'credentialRevocationId': self.credential_revocation_id,
             'revocationRegistryId': self.revocation_registry_id

legal-api/src/legal_api/resources/v2/business/business_digital_credentials.py

@@ -16,11 +16,19 @@
 from datetime import datetime
 from http import HTTPStatus
 
-from flask import Blueprint, current_app, jsonify, request
+from flask import Blueprint, _request_ctx_stack, current_app, jsonify, request
 from flask_cors import cross_origin
 
 from legal_api.extensions import socketio
-from legal_api.models import Business, DCConnection, DCDefinition, DCIssuedCredential
+from legal_api.models import (
+    Business,
+    CorpType,
+    DCConnection,
+    DCDefinition,
+    DCIssuedBusinessUserCredential,
+    DCIssuedCredential,
+    User,
+)
 from legal_api.services import digital_credentials
 from legal_api.utils.auth import jwt
 
@@ -82,10 +90,28 @@ def get_connections(identifier):
     return jsonify({'connections': response}), HTTPStatus.OK
 
 
-@bp.route('/<string:identifier>/digitalCredentials/connection', methods=['DELETE'], strict_slashes=False)
+@bp.route('/<string:identifier>/digitalCredentials/connections/<string:connection_id>',
+          methods=['DELETE'], strict_slashes=False)
 @cross_origin(origin='*')
 @jwt.requires_auth
-def delete_connection(identifier):
+def delete_connection(identifier, connection_id):
+    """Delete a connection."""
+    business = Business.find_by_identifier(identifier)
+    if not business:
+        return jsonify({'message': f'{identifier} not found.'}), HTTPStatus.NOT_FOUND
+
+    connection = DCConnection.find_by_connection_id(connection_id=connection_id)
+    if not connection:
+        return jsonify({'message': f'{identifier} connection not found.'}), HTTPStatus.NOT_FOUND
+
+    connection.delete()
+    return jsonify({'message': 'Connection has been deleted.'}), HTTPStatus.OK
+
+
+@bp.route('/<string:identifier>/digitalCredentials/activeConnection', methods=['DELETE'], strict_slashes=False)
+@cross_origin(origin='*')
+@jwt.requires_auth
+def delete_active_connection(identifier):
     """Delete an active connection for this business."""
     business = Business.find_by_identifier(identifier)
     if not business:
@@ -139,18 +165,27 @@ def send_credential(identifier, credential_type):
     if not business:
         return jsonify({'message': f'{identifier} not found'}), HTTPStatus.NOT_FOUND
 
+    user = User.find_by_jwt_token(_request_ctx_stack.top.current_user)
+    if not user:
+        return jsonify({'message': 'User not found'}, HTTPStatus.NOT_FOUND)
+
     connection = DCConnection.find_active_by(business_id=business.id)
-    definition = DCDefinition.find_by_credential_type(DCDefinition.CredentialType[credential_type])
+    definition = DCDefinition.find_by(DCDefinition.CredentialType[credential_type],
+                                      digital_credentials.business_schema_id,
+                                      digital_credentials.business_cred_def_id)
 
     issued_credentials = DCIssuedCredential.find_by(dc_connection_id=connection.id,
                                                     dc_definition_id=definition.id)
     if issued_credentials and issued_credentials[0].credential_exchange_id:
         return jsonify({'message': 'Already requested to issue credential.'}), HTTPStatus.INTERNAL_SERVER_ERROR
 
+    credential_data = _get_data_for_credential(definition.credential_type, business, user)
+    credential_id = next((item['value'] for item in credential_data if item['name'] == 'credential_id'), None)
+
     response = digital_credentials.issue_credential(
         connection_id=connection.connection_id,
         definition=definition,
-        data=_get_data_for_credential(definition.credential_type, business)
+        data=credential_data
     )
     if not response:
         return jsonify({'message': 'Failed to issue credential.'}), HTTPStatus.INTERNAL_SERVER_ERROR
@@ -159,12 +194,11 @@ def send_credential(identifier, credential_type):
         dc_definition_id=definition.id,
         dc_connection_id=connection.id,
         credential_exchange_id=response['cred_ex_id'],
-        # TODO: Add a real ID
-        credential_id='123456'
+        credential_id=credential_id
     )
     issued_credential.save()
 
-    return jsonify({'message': 'Credential offer has been sent.'}), HTTPStatus.OK
+    return jsonify(issued_credential.json), HTTPStatus.OK
 
 
 @bp.route('/<string:identifier>/digitalCredentials/<string:credential_id>/revoke',
@@ -181,8 +215,7 @@ def revoke_credential(identifier, credential_id):
     if not connection:
         return jsonify({'message': f'{identifier} active connection not found.'}), HTTPStatus.NOT_FOUND
 
-    # TODO: Use a real ID
-    issued_credential = DCIssuedCredential.find_by_credential_id(credential_id='123456')
+    issued_credential = DCIssuedCredential.find_by_credential_id(credential_id=credential_id)
     if not issued_credential or issued_credential.is_revoked:
         return jsonify({'message': f'{identifier} issued credential not found.'}), HTTPStatus.NOT_FOUND
 
@@ -206,8 +239,7 @@ def delete_credential(identifier, credential_id):
     if not business:
         return jsonify({'message': f'{identifier} not found.'}), HTTPStatus.NOT_FOUND
 
-    # TODO: Use a real ID
-    issued_credential = DCIssuedCredential.find_by_credential_id(credential_id='123456')
+    issued_credential = DCIssuedCredential.find_by_credential_id(credential_id=credential_id)
     if not issued_credential:
         return jsonify({'message': f'{identifier} issued credential not found.'}), HTTPStatus.NOT_FOUND
 
@@ -241,7 +273,6 @@ def webhook_notification(topic_name: str):
                 issued_credential.revocation_registry_id = json_input['rev_reg_id']
                 issued_credential.save()
         elif topic_name == 'issue_credential_v2_0':
-            # TODO: We want to deactivate the connection once the credential is issued
             issued_credential = DCIssuedCredential.find_by_credential_exchange_id(json_input['cred_ex_id'])
             if issued_credential and json_input['state'] == 'done':
                 issued_credential.date_of_issue = datetime.utcnow()
@@ -255,48 +286,76 @@ def webhook_notification(topic_name: str):
     return jsonify({'message': 'Webhook received.'}), HTTPStatus.OK
 
 
-def _get_data_for_credential(credential_type: DCDefinition.CredentialType, business: Business):
+def _get_data_for_credential(credential_type: DCDefinition.CredentialType, business: Business, user: User):
     if credential_type == DCDefinition.CredentialType.business:
+
+        # Find the credential id from dc_issued_business_user_credentials and if there isn't one create one
+        issued_business_user_credential = DCIssuedBusinessUserCredential.find_by(
+            business_id=business.id, user_id=user.id)
+        if not issued_business_user_credential:
+            issued_business_user_credential = DCIssuedBusinessUserCredential(business_id=business.id, user_id=user.id)
+            issued_business_user_credential.save()
+
+        credential_id = f'{issued_business_user_credential.id:08}'
+
+        business_type = CorpType.find_by_id(business.legal_type)
+        if business_type:
+            business_type = business_type.full_desc
+        else:
+            business_type = business.legal_type
+
+        registered_on_dateint = ''
+        if business.founding_date:
+            registered_on_dateint = business.founding_date.strftime('%Y%m%d')
+
+        company_status = Business.State(business.state).name
+
+        family_name = (user.lastname or '').upper()
+
+        given_names = (user.firstname + (' ' + user.middlename if user.middlename else '') or '').upper()
+
+        roles = ', '.join([party_role.role.title() for party_role in business.party_roles.all() if party_role.role])
+
         return [
             {
                 'name': 'credential_id',
-                'value': ''
+                'value':  credential_id or ''
             },
             {
                 'name': 'identifier',
-                'value': business.identifier
+                'value': business.identifier or ''
             },
             {
                 'name': 'business_name',
-                'value': business.legal_name
+                'value': business.legal_name or ''
             },
             {
                 'name': 'business_type',
-                'value': business.legal_type
+                'value': business_type or ''
             },
             {
                 'name': 'cra_business_number',
                 'value': business.tax_id or ''
             },
             {
                 'name': 'registered_on_dateint',
-                'value': business.founding_date.isoformat()
+                'value': registered_on_dateint or ''
             },
             {
                 'name': 'company_status',
-                'value': business.state
+                'value': company_status or ''
             },
             {
                 'name': 'family_name',
-                'value': ''
+                'value': family_name or ''
             },
             {
                 'name': 'given_names',
-                'value': ''
+                'value': given_names or ''
             },
             {
                 'name': 'role',
-                'value': ''
+                'value': roles or ''
             }
         ]
 

legal-api/tests/unit/resources/v2/test_business_digital_credentials.py

@@ -21,7 +21,7 @@
 from unittest.mock import patch
 
 from legal_api.services.authz import BASIC_USER
-from legal_api.models import DCDefinition
+from legal_api.models import DCDefinition, User
 from legal_api.services.digital_credentials import DigitalCredentialsService
 
 from tests.unit.models import factory_business
@@ -89,17 +89,20 @@ def test_send_credential(session, client, jwt):  # pylint:disable=unused-argumen
     headers = create_header(jwt, [BASIC_USER])
     identifier = 'FM1234567'
     business = factory_business(identifier)
-
-    create_dc_definition()
+    definition = create_dc_definition()
+    test_user = User(username='test-user', firstname='test', lastname='test')
+    test_user.save()
     create_dc_connection(business, is_active=True)
+    cred_ex_id = '3fa85f64-5717-4562-b3fc-2c963f66afa6'
 
-    with patch.object(DigitalCredentialsService, 'issue_credential', return_value={
-            'cred_ex_id': '3fa85f64-5717-4562-b3fc-2c963f66afa6'}):
-        rv = client.post(
-            f'/api/v2/businesses/{identifier}/digitalCredentials/{DCDefinition.CredentialType.business.name}',
-            headers=headers, content_type=content_type)
-        assert rv.status_code == HTTPStatus.OK
-        assert rv.json.get('message') == 'Credential offer has been sent.'
+    with patch.object(User, 'find_by_jwt_token', return_value=test_user):
+        with patch.object(DCDefinition, 'find_by', return_value=definition):
+            with patch.object(DigitalCredentialsService, 'issue_credential', return_value={'cred_ex_id': cred_ex_id}):
+                rv = client.post(
+                    f'/api/v2/businesses/{identifier}/digitalCredentials/{DCDefinition.CredentialType.business.name}',
+                    headers=headers, content_type=content_type)
+                assert rv.status_code == HTTPStatus.OK
+                assert rv.json.get('credentialExchangeId') == cred_ex_id
 
 
 def test_get_issued_credentials(session, client, jwt):  # pylint:disable=unused-argument

legal-api/tests/unit/services/test_digital_credentials.py

@@ -31,7 +31,7 @@ def test_init_app(session, app):  # pylint:disable=unused-argument
             digital_credentials.init_app(app)
             definition = DCDefinition.find_by_credential_type(DCDefinition.CredentialType.business)
             assert definition.schema_id == schema_id
-            assert definition.schema_name == digital_credentials.business_schema['schema_name']
-            assert definition.schema_version == digital_credentials.business_schema['schema_version']
+            assert definition.schema_name == digital_credentials.business_schema_name
+            assert definition.schema_version == digital_credentials.business_schema_version
             assert definition.credential_definition_id == cred_def_id
             assert not definition.is_deleted