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=_get_data_for_credential(definition.credential_type, business, user)
  )
  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/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