sign snapschot credential from webhook

This commit is contained in:
Cayo Puigdefabregas 2024-11-25 20:18:47 +01:00
parent 9ebd3b18a4
commit baa4d87a11
7 changed files with 75 additions and 22 deletions

View file

@ -105,6 +105,7 @@ class Command(BaseCommand):
assert dname
assert title
except Exception:
ldata = {}
title = ''
_name = ''

View file

@ -33,10 +33,11 @@ class UserView(LoginRequiredMixin):
]
def get(self, request, *args, **kwargs):
err_txt = "User domain is {} which does not match server domain {}".format(
request.get_host(), settings.DOMAIN
)
assert request.get_host() == settings.DOMAIN, err_txt
if not settings.DEVELOPMENT:
err_txt = "User domain is {} which does not match server domain {}".format(
request.get_host(), settings.DOMAIN
)
assert request.get_host() == settings.DOMAIN, err_txt
self.admin_validated = cache.get("KEY_DIDS")
response = super().get(request, *args, **kwargs)
@ -55,10 +56,11 @@ class UserView(LoginRequiredMixin):
return url or response
def post(self, request, *args, **kwargs):
err_txt = "User domain is {} which does not match server domain {}".format(
request.get_host(), settings.DOMAIN
)
assert request.get_host() == settings.DOMAIN, err_txt
if not settings.DEVELOPMENT:
err_txt = "User domain is {} which does not match server domain {}".format(
request.get_host(), settings.DOMAIN
)
assert request.get_host() == settings.DOMAIN, err_txt
self.admin_validated = cache.get("KEY_DIDS")
response = super().post(request, *args, **kwargs)
url = self.check_gdpr()

View file

@ -1,13 +1,12 @@
{
"@context": [
"https://www.w3.org/2018/credentials/v1",
"https://idhub.pangea.org/vc_schemas/v1/device-snapshot.json"
"https://www.w3.org/2018/credentials/v1"
],
"type": ["VerifiableCredential", "DeviceSnapshot"],
"issuer": "{{ issuer_did }}",
"issuanceDate": "{{ issuance_date }}",
"credentialSubject": {
"operatorId": "{{ operatorid }}",
"operatorId": "123456789011121314",
"uuid": "{{ uuid }}",
"type": "hardwareList",
"software": "workbench-script",
@ -44,24 +43,19 @@
},
{
"type": "HardwareList",
"operation": "{{ smartctl }}",
"operation": "smartctl",
"output": "{{ smartctl }}",
"timestamp": "{{ issuance_date }}"
},
{
"type": "HardwareList",
"operation": "{{ inxi }}",
"operation": "inxi",
"output": "{{ inxi }}",
"timestamp": "{{ issuance_date }}"
}
],
"credentialStatus": {
"id": "{{ credential_status_id}}",
"type": "RevocationBitmap2022",
"revocationBitmapIndex": "{{ id_credential }}"
},
"credentialSchema": {
"id": "https://idhub.pangea.org/vc_schemas/device-snapshot.json",
"id": "https://idhub.pangea.org/vc_schemas/device-snapshot-v1.json",
"type": "FullJsonSchemaValidator2021"
}
}

View file

@ -1,7 +1,7 @@
{
"$id": "https://idhub.pangea.org/vc_schemas/v1/device-snapshot.json",
"$id": "https://idhub.pangea.org/vc_schemas/device-snapshot-v1.json",
"$schema": "https://json-schema.org/draft/2020-12/schema",
"title": "DeviceSnapshot",
"title": "DeviceSnapshotV1",
"description": "Snapshot create by workbench-script, software for discover hardware in one device.",
"name": [
{

View file

@ -31,6 +31,7 @@ SECRET_KEY = config('SECRET_KEY')
# SECURITY WARNING: don't run with debug turned on in production!
DEBUG = config('DEBUG', default=False, cast=bool)
DEVELOPMENT = config('DEVELOPMENT', default=False, cast=bool)
DOMAIN = config("DOMAIN")
assert DOMAIN not in [None, ''], "DOMAIN var is MANDATORY"

View file

@ -7,6 +7,7 @@ app_name = 'webhook'
urlpatterns = [
path('verify/', views.webhook_verify, name='verify'),
path('sign/', views.webhook_issue, name='sign'),
path('tokens/', views.WebHookTokenView.as_view(), name='tokens'),
path('tokens/new', views.TokenNewView.as_view(), name='new_token'),
path('tokens/<int:pk>/del', views.TokenDeleteView.as_view(), name='delete_token'),

View file

@ -11,6 +11,8 @@ from pyvckit.verify import verify_vp, verify_vc
from uuid import uuid4
from idhub.mixins import AdminView
from idhub_auth.models import User
from idhub.models import DID, Schemas, VerificableCredential
from webhook.models import Token
from webhook.tables import TokensTable
@ -51,6 +53,59 @@ def webhook_verify(request):
return JsonResponse({'error': 'Invalid request method'}, status=400)
@csrf_exempt
def webhook_issue(request):
if request.method == 'POST':
auth_header = request.headers.get('Authorization')
if not auth_header or not auth_header.startswith('Bearer '):
return JsonResponse({'error': 'Invalid or missing token'}, status=401)
token = auth_header.split(' ')[1]
tk = Token.objects.filter(token=token).first()
if not tk:
return JsonResponse({'error': 'Invalid or missing token'}, status=401)
try:
data = json.loads(request.body)
except json.JSONDecodeError:
return JsonResponse({'error': 'Invalid JSON'}, status=400)
typ = data.get("type")
vc = data.get("data")
try:
vc = json.dumps(vc)
except Exception:
return JsonResponse({'error': 'Invalid JSON'}, status=400)
user = User.objects.filter(email=data.get("user")).first()
if not typ or not vc or not user:
return JsonResponse({'error': 'Invalid JSON'}, status=400)
did = DID.objects.filter(user__isnull=True).first()
if not did:
return JsonResponse({'error': 'Invalid DID'}, status=400)
schema = Schemas.objects.filter(file_schema=typ).first()
if not schema:
return JsonResponse({'error': 'Invalid credential'}, status=400)
cred = VerificableCredential(
csv_data=vc,
issuer_did=did,
schema=schema,
user=user
)
cred.set_type()
vc_signed = cred.issue(did, domain=request.get_host(), encrypt=False)
return JsonResponse({'status': 'success', "data": vc_signed}, status=200)
return JsonResponse({'status': 'fail'}, status=200)
return JsonResponse({'error': 'Invalid request method'}, status=400)
class WebHookTokenView(AdminView, SingleTableView):
template_name = "token.html"
title = _("Credential management")
@ -93,4 +148,3 @@ class TokenNewView(AdminView, View):
Token.objects.create(token=uuid4())
return redirect('webhook:tokens')