IdHub/idhub/admin/forms.py

454 lines
13 KiB
Python
Raw Normal View History

2023-11-08 10:24:37 +00:00
import json
2024-02-15 17:59:57 +00:00
import base64
import jsonschema
2023-11-08 10:24:37 +00:00
import pandas as pd
from nacl.exceptions import CryptoError
2023-10-30 12:53:19 +00:00
from django import forms
2024-01-18 14:09:19 +00:00
from django.core.cache import cache
2023-11-13 17:09:37 +00:00
from django.utils.translation import gettext_lazy as _
2023-11-08 10:24:37 +00:00
from django.core.exceptions import ValidationError
from utils import certs
2023-11-08 10:24:37 +00:00
from idhub.models import (
DID,
2023-11-13 17:09:37 +00:00
File_datas,
Membership,
2023-11-08 10:24:37 +00:00
Schemas,
2023-11-14 14:55:17 +00:00
UserRol,
2023-11-08 10:24:37 +00:00
VerificableCredential,
)
from idhub_auth.models import User
2023-10-30 12:53:19 +00:00
2024-02-05 11:25:02 +00:00
class TermsConditionsForm2(forms.Form):
accept = forms.BooleanField(
label=_("Accept terms and conditions of the service"),
required=False
)
def __init__(self, *args, **kwargs):
self.user = kwargs.pop('user', None)
super().__init__(*args, **kwargs)
def clean(self):
data = self.cleaned_data
if data.get("accept"):
self.user.accept_gdpr = True
else:
self.user.accept_gdpr = False
return data
def save(self, commit=True):
2024-02-05 11:25:02 +00:00
if commit:
self.user.save()
return self.user
return
class EncryptionKeyForm(forms.Form):
key = forms.CharField(
label=_("Key for encrypt the secrets of all system"),
required=True
)
def clean(self):
data = self.cleaned_data
self._key = data["key"]
if not DID.objects.exists():
return data
did = DID.objects.first()
cache.set("KEY_DIDS", self._key, None)
try:
did.get_key_material()
except CryptoError:
cache.set("KEY_DIDS", None)
txt = _("Key no valid!")
raise ValidationError(txt)
2024-02-21 09:46:33 +00:00
cache.set("KEY_DIDS", None)
return data
def save(self, commit=True):
if commit:
cache.set("KEY_DIDS", self._key, None)
2024-02-23 18:17:56 +00:00
if not DID.objects.exists():
did = DID.objects.create(label='Default', type=DID.Types.WEB)
did.set_did()
did.save()
return
2024-02-05 11:25:02 +00:00
class TermsConditionsForm(forms.Form):
accept_privacy = forms.BooleanField(
widget=forms.CheckboxInput(attrs={'class': 'form-check-input'}),
required=False
)
accept_legal = forms.BooleanField(
widget=forms.CheckboxInput(attrs={'class': 'form-check-input'}),
required=False
)
accept_cookies = forms.BooleanField(
widget=forms.CheckboxInput(attrs={'class': 'form-check-input'}),
required=False
)
def __init__(self, *args, **kwargs):
self.user = kwargs.pop('user', None)
super().__init__(*args, **kwargs)
def get_label(self, url, read):
label = _('I read and accepted the')
label += f' <a class="btn btn-green-admin" target="_blank" href="{url}" '
label += f'title="{read}">{read}</a>'
return label
def privacy_label(self):
url = "https://laweb.pangea.org/politica-de-privacitat/"
2024-02-05 13:58:54 +00:00
read = _("Privacy policy")
2024-02-05 11:25:02 +00:00
return self.get_label(url, read)
def legal_label(self):
url = "https://laweb.pangea.org/avis-legal/"
2024-02-05 13:58:54 +00:00
read = _("Legal policy")
2024-02-05 11:25:02 +00:00
return self.get_label(url, read)
def cookies_label(self):
url = "https://laweb.pangea.org/politica-de-cookies-2/"
2024-02-05 13:58:54 +00:00
read = _("Cookies policy")
2024-02-05 11:25:02 +00:00
return self.get_label(url, read)
def clean(self):
data = self.cleaned_data
privacy = data.get("accept_privacy")
legal = data.get("accept_legal")
cookies = data.get("accept_cookies")
if privacy and legal and cookies:
self.user.accept_gdpr = True
else:
self.user.accept_gdpr = False
return data
def save(self, commit=True):
if commit:
self.user.save()
return self.user
return
2023-10-30 12:53:19 +00:00
class ImportForm(forms.Form):
2023-12-14 16:15:22 +00:00
did = forms.ChoiceField(label=_("Did"), choices=[])
2024-01-18 14:09:19 +00:00
eidas1 = forms.ChoiceField(
label=_("Signature with Eidas1"),
choices=[],
required=False
)
2023-12-14 16:15:22 +00:00
schema = forms.ChoiceField(label=_("Schema"), choices=[])
file_import = forms.FileField(label=_("File import"))
2023-11-02 09:59:15 +00:00
2023-11-08 10:24:37 +00:00
def __init__(self, *args, **kwargs):
self._schema = None
self._did = None
2024-01-18 14:09:19 +00:00
self._eidas1 = None
2023-11-08 10:24:37 +00:00
self.rows = {}
2023-11-22 10:18:47 +00:00
self.properties = {}
self.users = []
2023-11-08 10:24:37 +00:00
super().__init__(*args, **kwargs)
2024-02-23 18:17:56 +00:00
dids = DID.objects.filter(user__isnull=True)
2023-11-08 10:24:37 +00:00
self.fields['did'].choices = [
2024-01-18 14:09:19 +00:00
(x.did, x.label) for x in dids.filter(eidas1=False)
2023-11-08 10:24:37 +00:00
]
self.fields['schema'].choices = [(0, _('Select one'))] + [
(x.id, x.name) for x in Schemas.objects.filter()
2023-11-08 10:24:37 +00:00
]
2024-01-18 14:09:19 +00:00
if dids.filter(eidas1=True).exists():
choices = [("", "")]
choices.extend([
(x.did, x.label) for x in dids.filter(eidas1=True)
])
self.fields['eidas1'].choices = choices
else:
self.fields.pop('eidas1')
2023-11-08 10:24:37 +00:00
2024-01-18 14:09:19 +00:00
def clean(self):
2023-11-08 10:24:37 +00:00
data = self.cleaned_data["did"]
did = DID.objects.filter(
2024-02-23 18:17:56 +00:00
user__isnull=True,
2023-11-08 10:24:37 +00:00
did=data
)
if not did.exists():
raise ValidationError("Did is not valid!")
self._did = did.first()
2024-01-18 14:09:19 +00:00
eidas1 = self.cleaned_data.get('eidas1')
if eidas1:
self._eidas1 = DID.objects.filter(
2024-02-23 18:17:56 +00:00
user__isnull=True,
2024-01-18 14:09:19 +00:00
eidas1=True,
did=eidas1
).first()
2023-11-08 10:24:37 +00:00
return data
def clean_schema(self):
data = self.cleaned_data["schema"]
schema = Schemas.objects.filter(
id=data
)
if not schema.exists():
raise ValidationError("Schema is not valid!")
self._schema = schema.first()
2023-11-22 10:18:47 +00:00
try:
2024-02-15 17:59:57 +00:00
self.json_schema = self._schema.get_credential_subject_schema()
2023-11-22 10:18:47 +00:00
except Exception:
raise ValidationError("Schema is not valid!")
2023-11-08 10:24:37 +00:00
return data
def clean_file_import(self):
data = self.cleaned_data["file_import"]
self.file_name = data.name
2024-01-10 17:35:12 +00:00
df = pd.read_excel(data)
2024-02-15 17:59:57 +00:00
# convert dates to iso 8601
for col in df.select_dtypes(include='datetime').columns:
df[col] = df[col].dt.strftime("%Y-%m-%d")
2023-11-08 10:24:37 +00:00
data_pd = df.fillna('').to_dict()
if not data_pd:
self.exception("This file is empty!")
if not self._schema:
return data
2023-11-08 10:24:37 +00:00
for n in range(df.last_valid_index()+1):
row = {}
2023-11-08 10:24:37 +00:00
for k in data_pd.keys():
2024-02-15 17:59:57 +00:00
row[k] = data_pd[k][n] or ''
2023-11-08 10:24:37 +00:00
user = self.validate_jsonld(n, row)
self.rows[user] = row
return data
def save(self, commit=True):
table = []
for k, v in self.rows.items():
table.append(self.create_credential(k, v))
if commit:
for cred in table:
cred.save()
File_datas.objects.create(file_name=self.file_name)
return table
return
def validate_jsonld(self, line, row):
try:
2024-02-26 15:00:44 +00:00
jsonschema.validate(
instance=row,
schema=self.json_schema,
format_checker=jsonschema.Draft202012Validator.FORMAT_CHECKER
)
2024-02-15 17:59:57 +00:00
except jsonschema.exceptions.ValidationError as err:
msg = "line {}: {}".format(line+1, err)
return self.exception(msg)
2023-11-08 10:24:37 +00:00
user, new = User.objects.get_or_create(email=row.get('email'))
if new:
self.users.append(user)
2024-02-22 12:50:07 +00:00
user.set_encrypted_sensitive_data()
user.save()
self.create_defaults_dids(user)
2023-11-08 10:24:37 +00:00
return user
2023-11-08 10:24:37 +00:00
2024-02-22 12:50:07 +00:00
def create_defaults_dids(self, user):
did = DID(label="Default", user=user, type=DID.Types.WEB)
did.set_did()
did.save()
2023-11-08 10:24:37 +00:00
def create_credential(self, user, row):
bcred = VerificableCredential.objects.filter(
user=user,
schema=self._schema,
issuer_did=self._did,
status=VerificableCredential.Status.ENABLED
)
if bcred.exists():
cred = bcred.first()
cred.csv_data = json.dumps(row, default=str)
cred.eidas1_did = self._eidas1
return cred
2024-01-24 15:10:10 +00:00
cred = VerificableCredential(
2023-11-08 10:24:37 +00:00
verified=False,
user=user,
2024-02-02 17:01:59 +00:00
csv_data=json.dumps(row, default=str),
2023-11-15 10:43:13 +00:00
issuer_did=self._did,
2023-11-14 17:50:23 +00:00
schema=self._schema,
2024-01-18 14:09:19 +00:00
eidas1_did=self._eidas1
2023-11-08 10:24:37 +00:00
)
2024-01-24 15:10:10 +00:00
cred.set_type()
return cred
2023-11-08 10:24:37 +00:00
def exception(self, msg):
File_datas.objects.create(file_name=self.file_name, success=False)
raise ValidationError(msg)
2023-11-02 09:59:15 +00:00
class SchemaForm(forms.Form):
2023-12-14 16:59:40 +00:00
file_template = forms.FileField(label=_("File template"))
2023-11-13 17:09:37 +00:00
class MembershipForm(forms.ModelForm):
class Meta:
model = Membership
fields = ['type', 'start_date', 'end_date']
def clean_end_date(self):
data = super().clean()
start_date = data['start_date']
end_date = data.get('end_date')
2023-11-14 14:55:17 +00:00
members = Membership.objects.filter(
type=data['type'],
user=self.instance.user
)
if self.instance.id:
members = members.exclude(id=self.instance.id)
2023-11-16 13:17:19 +00:00
if members.filter(start_date__lte=start_date, end_date=None).exists():
msg = _("This membership already exists!")
raise forms.ValidationError(msg)
2023-11-13 17:09:37 +00:00
if (start_date and end_date):
if start_date > end_date:
msg = _("The end date is less than the start date")
raise forms.ValidationError(msg)
2023-11-14 14:55:17 +00:00
members = members.filter(
2023-11-13 17:09:37 +00:00
start_date__lte=end_date,
end_date__gte=start_date,
)
2023-11-14 14:55:17 +00:00
if members.exists():
2023-11-13 17:09:37 +00:00
msg = _("This membership already exists!")
raise forms.ValidationError(msg)
2023-11-16 13:17:19 +00:00
if not end_date:
members = members.filter(
start_date__gte=start_date,
)
if members.exists():
msg = _("This membership already exists!")
raise forms.ValidationError(msg)
2023-11-13 17:09:37 +00:00
return end_date
2023-11-14 14:55:17 +00:00
class UserRolForm(forms.ModelForm):
class Meta:
model = UserRol
fields = ['service']
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
if not self.instance.id:
user = self.instance.user
choices = self.fields['service'].choices
choices.queryset = choices.queryset.exclude(users__user=user)
self.fields['service'].choices = choices
def clean_service(self):
data = super().clean()
service = UserRol.objects.filter(
service=data['service'],
user=self.instance.user
)
if service.exists():
msg = _("Is not possible to have a duplicate role")
raise forms.ValidationError(msg)
return data['service']
2024-01-10 12:53:43 +00:00
class ImportCertificateForm(forms.Form):
label = forms.CharField(label=_("Label"))
password = forms.CharField(
label=_("Password of certificate"),
widget=forms.PasswordInput
)
file_import = forms.FileField(label=_("File import"))
def __init__(self, *args, **kwargs):
self._did = None
self._s = None
self._label = None
self.user = kwargs.pop('user', None)
super().__init__(*args, **kwargs)
def clean(self):
data = super().clean()
file_import = data.get('file_import')
self.pfx_file = file_import.read()
self.file_name = file_import.name
self._pss = data.get('password')
self._label = data.get('label')
if not self.pfx_file or not self._pss:
msg = _("Is not a valid certificate")
raise forms.ValidationError(msg)
self.signer_init()
if not self._s:
msg = _("Is not a valid certificate")
raise forms.ValidationError(msg)
self.new_did()
return data
def new_did(self):
keys = {
"cert": base64.b64encode(self.pfx_file).decode('utf-8'),
"passphrase": self._pss
}
key_material = json.dumps(keys)
self._did = DID(
key_material=key_material,
did=self.file_name,
label=self._label,
eidas1=True,
2024-01-18 14:09:19 +00:00
user=self.user,
type=DID.Types.KEY
2024-01-10 12:53:43 +00:00
)
2024-02-21 09:46:33 +00:00
self._did.set_key_material(key_material)
2024-01-18 14:09:19 +00:00
2024-01-10 12:53:43 +00:00
def save(self, commit=True):
if commit:
self._did.save()
return self._did
return
def signer_init(self):
self._s = certs.load_cert(
self.pfx_file, self._pss.encode('utf-8')
)