import json import requests import secrets import nacl import base64 from nacl import pwhash, secret from django.core.cache import cache from django.conf import settings from django.http import QueryDict from django.utils.translation import gettext_lazy as _ from idhub_auth.models import User from django.db import models from utils.idhub_ssikit import verify_presentation SALT_CHARS = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" def gen_salt(length: int) -> str: """Generate a random string of SALT_CHARS with specified ``length``.""" if length <= 0: raise ValueError("Salt length must be positive") return "".join(secrets.choice(SALT_CHARS) for _ in range(length)) def set_client_id(): return gen_salt(24) def set_client_secret(): return gen_salt(48) def set_code(): return gen_salt(24) class Organization(models.Model): """ This class represent a member of one net trust or federated host. Client_id and client_secret are the credentials of this organization get a connection to my. (receive a request) My_client_id and my_client_secret are my credentials than to use if I want to connect to this organization. (send a request) For use the packages requests we need use my_client_id For use in the get or post method of a View, then we need use client_id and secret_id """ name = models.CharField(max_length=250) client_id = models.CharField( max_length=24, default=set_client_id, unique=True ) client_secret = models.CharField( max_length=48, default=set_client_secret ) my_client_id = models.CharField( max_length=24, ) my_client_secret = models.CharField( max_length=48, ) response_uri = models.URLField( help_text=_("Url where to send the verificable presentation"), max_length=250 ) encrypted_sensitive_data = models.CharField(max_length=255, default=None, null=True) salt = models.CharField(max_length=255, default=None, null=True) def send(self, vp, code): """ Send the verificable presentation to Verifier """ url = "{url}/verify".format( url=self.response_uri.strip("/"), ) auth = (self.my_client_id, self.my_client_secret) data = {"vp_token": vp} if code: data["code"] = code return requests.post(url, data=data, auth=auth) def demand_authorization(self): """ Send the a request for start a process of Verifier """ url = "{url}/verify?demand_uri={redirect_uri}".format( url=self.response_uri.strip("/"), redirect_uri=settings.RESPONSE_URI ) auth = (self.my_client_id, self.my_client_secret) return requests.get(url, auth=auth) def derive_key_from_password(self, password=None): if not password: password = cache.get("KEY_DIDS").encode('utf-8') kdf = pwhash.argon2i.kdf ops = pwhash.argon2i.OPSLIMIT_INTERACTIVE mem = pwhash.argon2i.MEMLIMIT_INTERACTIVE return kdf( secret.SecretBox.KEY_SIZE, password, self.get_salt(), opslimit=ops, memlimit=mem ) def decrypt_sensitive_data(self, data=None): sb_key = self.derive_key_from_password() sb = secret.SecretBox(sb_key) if not data: data = self.get_encrypted_sensitive_data() if not isinstance(data, bytes): data = data.encode('utf-8') return sb.decrypt(data).decode('utf-8') def encrypt_sensitive_data(self, data): sb_key = self.derive_key_from_password() sb = secret.SecretBox(sb_key) if not isinstance(data, bytes): data = data.encode('utf-8') return base64.b64encode(sb.encrypt(data)).decode('utf-8') def get_salt(self): if not self.salt: return '' return base64.b64decode(self.salt.encode('utf-8')) def set_salt(self): self.salt = base64.b64encode(nacl.utils.random(16)).decode('utf-8') def get_encrypted_sensitive_data(self): return base64.b64decode(self.encrypted_sensitive_data.encode('utf-8')) def set_encrypted_sensitive_data(self): key = base64.b64encode(nacl.utils.random(64)) self.set_salt() key_crypted = self.encrypt_sensitive_data(key) self.encrypted_sensitive_data = key_crypted def encrypt_data(self, data): pw = self.decrypt_sensitive_data() sb = self.get_secret_box(pw) value_enc = sb.encrypt(data.encode('utf-8')) return base64.b64encode(value_enc).decode('utf-8') def decrypt_data(self, data): pw = self.decrypt_sensitive_data() sb = self.get_secret_box(pw) value = base64.b64decode(data.encode('utf-8')) return sb.decrypt(value).decode('utf-8') def get_secret_box(self, password): sb_key = self.derive_key_from_password(password) return secret.SecretBox(sb_key) def change_password_key(self, new_password): data = self.decrypt_sensitive_data() sb_key = self.derive_key_from_password(new_password) sb = secret.SecretBox(sb_key) if not isinstance(data, bytes): data = data.encode('utf-8') encrypted_data = base64.b64encode(sb.encrypt(data)).decode('utf-8') self.encrypted_sensitive_data = encrypted_data def __str__(self): return self.name ################### # Verifier clases # ################### class Authorization(models.Model): """ This class represent a query through browser the client to the wallet. The Verifier need to do a redirection to the user to Wallet. The code we use as a soft foreing key between Authorization and OAuth2VPToken. """ code = models.CharField(max_length=24, default=set_code) code_used = models.BooleanField(default=False) created = models.DateTimeField(auto_now=True) presentation_definition = models.CharField(max_length=250) organization = models.ForeignKey( Organization, on_delete=models.CASCADE, related_name='authorizations', null=True, ) user = models.ForeignKey( User, on_delete=models.CASCADE, null=True, ) def authorize(self, path=None): data = { "response_type": "vp_token", "response_mode": "direct_post", "client_id": self.organization.my_client_id, "presentation_definition": self.presentation_definition, "code": self.code, "nonce": gen_salt(5), } query_dict = QueryDict('', mutable=True) query_dict.update(data) response_uri = self.organization.response_uri.strip("/") if path: response_uri = "{}/{}".format(response_uri, path.strip("/")) url = '{response_uri}/authorize?{params}'.format( response_uri=response_uri, params=query_dict.urlencode() ) return url class OAuth2VPToken(models.Model): """ This class represent the response of Wallet to Verifier and the result of verify. """ created = models.DateTimeField(auto_now=True) result_verify = models.CharField(max_length=255) vp_token = models.TextField() organization = models.ForeignKey( Organization, on_delete=models.CASCADE, related_name='vp_tokens', null=True, ) user = models.ForeignKey( User, on_delete=models.CASCADE, related_name='vp_tokens', null=True, ) authorization = models.ForeignKey( Authorization, on_delete=models.SET_NULL, related_name='vp_tokens', null=True, ) def __init__(self, *args, **kwargs): code = kwargs.pop("code", None) super().__init__(*args, **kwargs) self.authorization = Authorization.objects.filter(code=code).first() def verifing(self): self.result_verify = verify_presentation(self.vp_token) def get_response_verify(self): response = { "verify": ',', "redirect_uri": "", "response": "", } verification = json.loads(self.result_verify) if verification.get('errors') or verification.get('warnings'): response["verify"] = "Error, Verification Failed" return response response["verify"] = "Ok, Verification correct" url = self.get_redirect_url() if url: response["redirect_uri"] = url return response def get_redirect_url(self): if not settings.ALLOW_CODE_URI: return data = { "code": self.authorization.code, } query_dict = QueryDict('', mutable=True) query_dict.update(data) response_uri = settings.ALLOW_CODE_URI url = '{response_uri}?{params}'.format( response_uri=response_uri, params=query_dict.urlencode() ) return url def get_user_info(self): tk = json.loads(self.vp_token) self.user_info = tk.get( "verifiableCredential", [{}] )[-1].get("credentialSubject")