forked from enviPath/enviPy
wip
This commit is contained in:
Tim Lorsbach
@ -1,17 +1,22 @@
|
||||
import hashlib
|
||||
from collections import defaultdict
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
import jwt
|
||||
import requests
|
||||
|
||||
import nh3
|
||||
from django.conf import settings as s
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.http import HttpResponse, JsonResponse
|
||||
from django.shortcuts import redirect
|
||||
from ninja import Field, Form, Query, Router, Schema
|
||||
from ninja.errors import HttpError
|
||||
from ninja.security import HttpBearer
|
||||
from ninja.security import SessionAuth
|
||||
|
||||
from utilities.chem import FormatConverter
|
||||
from utilities.misc import PackageExporter
|
||||
|
||||
from .logic import (
|
||||
EPDBURLParser,
|
||||
GroupManager,
|
||||
@ -59,7 +64,46 @@ def _anonymous_or_real(request):
|
||||
return get_user_model().objects.get(username="anonymous")
|
||||
|
||||
|
||||
router = Router(auth=SessionAuth(csrf=False))
|
||||
def validate_token(token: str) -> dict:
|
||||
TENANT_ID = s.MS_ENTRA_TENANT_ID
|
||||
CLIENT_ID = s.MS_ENTRA_CLIENT_ID
|
||||
|
||||
# Fetch Microsoft's public keys
|
||||
jwks_uri = f"https://login.microsoftonline.com/{TENANT_ID}/discovery/v2.0/keys"
|
||||
jwks = requests.get(jwks_uri).json()
|
||||
|
||||
header = jwt.get_unverified_header(token)
|
||||
|
||||
public_key = jwt.algorithms.RSAAlgorithm.from_jwk(
|
||||
next(k for k in jwks["keys"] if k["kid"] == header["kid"])
|
||||
)
|
||||
|
||||
claims = jwt.decode(
|
||||
token,
|
||||
public_key,
|
||||
algorithms=["RS256"],
|
||||
audience=[CLIENT_ID, f"api://{CLIENT_ID}"],
|
||||
issuer=f"https://sts.windows.net/{TENANT_ID}/",
|
||||
)
|
||||
return claims
|
||||
|
||||
|
||||
class MSBearerTokenAuth(HttpBearer):
|
||||
|
||||
def authenticate(self, request, token):
|
||||
if token is None:
|
||||
return None
|
||||
|
||||
claims = validate_token(token)
|
||||
|
||||
if not User.objects.filter(uuid=claims['oid']).exists():
|
||||
return None
|
||||
|
||||
request.user = User.objects.get(uuid=claims['oid'])
|
||||
return request.user
|
||||
|
||||
|
||||
router = Router(auth=MSBearerTokenAuth())
|
||||
|
||||
|
||||
class Error(Schema):
|
||||
@ -153,59 +197,6 @@ class SimpleModel(SimpleObject):
|
||||
identifier: str = "relative-reasoning"
|
||||
|
||||
|
||||
################
|
||||
# Login/Logout #
|
||||
################
|
||||
@router.post("/", response={200: SimpleUser, 403: Error}, auth=None)
|
||||
def login(request, loginusername: Form[str], loginpassword: Form[str]):
|
||||
from django.contrib.auth import authenticate, login
|
||||
|
||||
if request.headers.get("Authorization"):
|
||||
import jwt
|
||||
import requests
|
||||
|
||||
TENANT_ID = s.MS_ENTRA_TENANT_ID
|
||||
CLIENT_ID = s.MS_ENTRA_CLIENT_ID
|
||||
|
||||
def validate_token(token: str) -> dict:
|
||||
# Fetch Microsoft's public keys
|
||||
jwks_uri = f"https://login.microsoftonline.com/{TENANT_ID}/discovery/v2.0/keys"
|
||||
jwks = requests.get(jwks_uri).json()
|
||||
|
||||
header = jwt.get_unverified_header(token)
|
||||
|
||||
public_key = jwt.algorithms.RSAAlgorithm.from_jwk(
|
||||
next(k for k in jwks["keys"] if k["kid"] == header["kid"])
|
||||
)
|
||||
|
||||
claims = jwt.decode(
|
||||
token,
|
||||
public_key,
|
||||
algorithms=["RS256"],
|
||||
audience=[CLIENT_ID, f"api://{CLIENT_ID}"],
|
||||
issuer=f"https://sts.windows.net/{TENANT_ID}/",
|
||||
)
|
||||
return claims
|
||||
|
||||
token = request.headers.get("Authorization").split(" ")[1]
|
||||
|
||||
claims = validate_token(token)
|
||||
|
||||
if not User.objects.filter(uuid=claims['oid']).exists():
|
||||
user = None
|
||||
else:
|
||||
user = User.objects.get(uuid=claims['oid'])
|
||||
|
||||
else:
|
||||
email = User.objects.get(username=loginusername).email
|
||||
user = authenticate(username=email, password=loginpassword)
|
||||
|
||||
if user:
|
||||
login(request, user)
|
||||
return user
|
||||
else:
|
||||
return 403, {"message": "Invalid username and/or password"}
|
||||
|
||||
|
||||
########
|
||||
# User #
|
||||
|
||||
@ -627,6 +627,25 @@ class PackageManager(object):
|
||||
else:
|
||||
pack.reviewed = False
|
||||
|
||||
# EDIT START
|
||||
if data.get("classification"):
|
||||
if data["classification"] == "INTERNAL":
|
||||
pack.classification = Package.Classification.RESTRICTED
|
||||
elif data["classification"] == "RESTRICTED":
|
||||
pack.classification = Package.Classification.RESTRICTED
|
||||
elif data["classification"] == "SECRET":
|
||||
pack.classification = Package.Classification.SECRET
|
||||
|
||||
if not "datapool" in data:
|
||||
raise ValueError("Missing datapool in package")
|
||||
|
||||
g = Group.objects.get(uuid=data["datapool"].split('/')[-1])
|
||||
pack.data_pool = g
|
||||
else:
|
||||
raise ValueError(f"Invalid classification {data['classification']}")
|
||||
|
||||
# EDIT END
|
||||
|
||||
pack.description = data["description"]
|
||||
pack.save()
|
||||
|
||||
@ -712,7 +731,13 @@ class PackageManager(object):
|
||||
default_structure = None
|
||||
|
||||
for structure in compound["structures"]:
|
||||
struc = CompoundStructure()
|
||||
if structure.get("pesLink"):
|
||||
from bayer.models import PESStructure
|
||||
struc = PESStructure()
|
||||
struc.pes_link = structure["pesLink"]
|
||||
else:
|
||||
struc = CompoundStructure()
|
||||
|
||||
# struc.object_url = Command.get_id(structure, keep_ids)
|
||||
struc.compound = comp
|
||||
struc.uuid = UUID(structure["id"].split("/")[-1]) if keep_ids else uuid4()
|
||||
@ -720,6 +745,10 @@ class PackageManager(object):
|
||||
struc.description = structure["description"]
|
||||
struc.aliases = structure.get("aliases", [])
|
||||
struc.smiles = structure["smiles"]
|
||||
|
||||
if structure.get("molfile"):
|
||||
struc.molfile = structure["molfile"]
|
||||
|
||||
struc.save()
|
||||
|
||||
for scen in structure["scenarios"]:
|
||||
|
||||
@ -1113,6 +1113,7 @@ class CompoundStructure(
|
||||
canonical_smiles = models.TextField(blank=False, null=False, verbose_name="Canonical SMILES")
|
||||
inchikey = models.TextField(max_length=27, blank=False, null=False, verbose_name="InChIKey")
|
||||
normalized_structure = models.BooleanField(null=False, blank=False, default=False)
|
||||
molfile = models.TextField(blank=True, null=True, verbose_name="Molfile")
|
||||
|
||||
external_identifiers = GenericRelation("ExternalIdentifier")
|
||||
|
||||
@ -1209,6 +1210,9 @@ class CompoundStructure(
|
||||
|
||||
return dict(hls)
|
||||
|
||||
def d3_json(self):
|
||||
return {}
|
||||
|
||||
|
||||
class EnzymeLink(EnviPathModel, KEGGIdentifierMixin):
|
||||
rule = models.ForeignKey("Rule", on_delete=models.CASCADE, db_index=True)
|
||||
@ -2215,7 +2219,9 @@ class Node(EnviPathModel, AliasMixin, ScenarioMixin, AdditionalInformationMixin)
|
||||
if isinstance(ai.get(), PropertyPrediction):
|
||||
predicted_properties[ai.get().__class__.__name__].append(ai.data)
|
||||
|
||||
return {
|
||||
extra_structure_data = self.default_node_label.d3_json()
|
||||
|
||||
res = {
|
||||
"depth": self.depth,
|
||||
"stereo_removed": self.stereo_removed,
|
||||
"url": self.url,
|
||||
@ -2224,6 +2230,7 @@ class Node(EnviPathModel, AliasMixin, ScenarioMixin, AdditionalInformationMixin)
|
||||
"image_svg": IndigoUtils.mol_to_svg(
|
||||
self.default_node_label.smiles, width=40, height=40
|
||||
),
|
||||
|
||||
"name": self.get_name(),
|
||||
"smiles": self.default_node_label.smiles,
|
||||
"scenarios": [{"name": s.get_name(), "url": s.url} for s in self.scenarios.all()],
|
||||
@ -2238,6 +2245,9 @@ class Node(EnviPathModel, AliasMixin, ScenarioMixin, AdditionalInformationMixin)
|
||||
"timeseries": self.get_timeseries_data(),
|
||||
}
|
||||
|
||||
res.update(**extra_structure_data)
|
||||
return res
|
||||
|
||||
@staticmethod
|
||||
@transaction.atomic
|
||||
def create(
|
||||
|
||||
Reference in New Issue
Block a user