From 61346c40979f39756b1a06298e9b52b0c85890e0 Mon Sep 17 00:00:00 2001 From: Liam Brydon <62733830+MyCreativityOutlet@users.noreply.github.com> Date: Tue, 21 Oct 2025 10:09:10 +1300 Subject: [PATCH] nh3 clean is now used on all free-text fields to ensure only approved html will be saved to the database. #72 --- epdb/templatetags/envipytags.py | 4 +- epdb/views.py | 170 ++++++++++++++++++-------------- 2 files changed, 101 insertions(+), 73 deletions(-) diff --git a/epdb/templatetags/envipytags.py b/epdb/templatetags/envipytags.py index 33346ff6..071946df 100644 --- a/epdb/templatetags/envipytags.py +++ b/epdb/templatetags/envipytags.py @@ -12,6 +12,8 @@ def classname(obj): @register.filter(name="nh_safe") -def nh_safe(txt: str): +def nh_safe(txt): + if not isinstance(txt, str): + return txt clean_html = nh3.clean(txt, tags=s.ALLOWED_HTML_TAGS) return mark_safe(clean_html) diff --git a/epdb/views.py b/epdb/views.py index 733488d8..d960ddab 100644 --- a/epdb/views.py +++ b/epdb/views.py @@ -58,14 +58,14 @@ def log_post_params(request): logger.debug(f"{k}\t{v}") -def sanitize_dict(bad_dict): +def clean_dict(bad_dict): """Check each value in a dictionary for XSS attempts""" - clean_dict = {} # TODO: I'm not sure if this is the best way to do this. + clean = {} # TODO: I'm not sure if this is the best way to do this. for key, value in bad_dict.items(): if key != 'csrfmiddlewaretoken' and isinstance(value, str): value = nh3.clean(value, tags=s.ALLOWED_HTML_TAGS) - clean_dict[key] = value - return clean_dict + clean[key] = value + return clean def error(request, message: str, detail: str, code: int = 400): @@ -400,11 +400,11 @@ def packages(request): else: return HttpResponseBadRequest() else: - package_name = request.POST.get("package-name") - # Sanitize for potential XSS + # Clean for potential XSS + package_name = nh3.clean(request.POST.get("package-name"), tags=s.ALLOWED_HTML_TAGS).strip() package_description = nh3.clean(request.POST.get( "package-description", s.DEFAULT_VALUES["description"] - ), tags=s.ALLOWED_HTML_TAGS) + ), tags=s.ALLOWED_HTML_TAGS).strip() created_package = PackageManager.create_package( current_user, package_name, package_description @@ -679,7 +679,9 @@ def search(request): if request.method == "GET": package_urls = request.GET.getlist("packages") - searchterm = request.GET.get("search") + # Clean for potential XSS + searchterm = nh3.clean(request.GET.get("search"), tags=s.ALLOWED_HTML_TAGS).strip() + mode = request.GET.get("mode") # add HTTP_ACCEPT check to differentiate between index and ajax call @@ -780,9 +782,9 @@ def package_models(request, package_uuid): elif request.method == "POST": log_post_params(request) - - name = request.POST.get("model-name") - description = request.POST.get("model-description") + # Clean for potential XSS + name = nh3.clean(request.POST.get("model-name"), tags=s.ALLOWED_HTML_TAGS).strip() + description = nh3.clean(request.POST.get("model-description"), tags=s.ALLOWED_HTML_TAGS).strip() model_type = request.POST.get("model-type") @@ -864,7 +866,8 @@ def package_model(request, package_uuid, model_uuid): ad_assessment = request.GET.get("app-domain-assessment", False) if classify or ad_assessment: - smiles = request.GET.get("smiles", "").strip() + # Clean for potential XSS + smiles = nh3.clean(request.GET.get("smiles", ""), tags=s.ALLOWED_HTML_TAGS).strip() # Check if smiles is non empty and valid if smiles == "": @@ -927,8 +930,9 @@ def package_model(request, package_uuid, model_uuid): else: return HttpResponseBadRequest() else: - name = request.POST.get("model-name", "").strip() - description = request.POST.get("model-description", "").strip() + # Clean for potential XSS + name = nh3.clean(request.POST.get("model-name", "").strip(), tags=s.ALLOWED_HTML_TAGS).strip() + description = nh3.clean(request.POST.get("model-description", "").strip(), tags=s.ALLOWED_HTML_TAGS).strip() if any([name, description]): if name: @@ -1030,9 +1034,9 @@ def package(request, package_uuid): else: return HttpResponseBadRequest() - new_package_name = request.POST.get("package-name") - # Sanitize for potential XSS - new_package_description = nh3.clean(request.POST.get("package-description"), tags=s.ALLOWED_HTML_TAGS) + # Clean for potential XSS + new_package_name = nh3.clean(request.POST.get("package-name"), tags=s.ALLOWED_HTML_TAGS).strip() + new_package_description = nh3.clean(request.POST.get("package-description"), tags=s.ALLOWED_HTML_TAGS).strip() grantee_url = request.POST.get("grantee") read = request.POST.get("read") == "on" @@ -1140,9 +1144,10 @@ def package_compounds(request, package_uuid): return render(request, "collections/objects_list.html", context) elif request.method == "POST": - compound_name = request.POST.get("compound-name") - compound_smiles = request.POST.get("compound-smiles") - compound_description = request.POST.get("compound-description") + # Clean for potential XSS + compound_name = nh3.clean(request.POST.get("compound-name"), tags=s.ALLOWED_HTML_TAGS).strip() + compound_smiles = nh3.clean(request.POST.get("compound-smiles"), tags=s.ALLOWED_HTML_TAGS).strip() + compound_description = nh3.clean(request.POST.get("compound-description"), tags=s.ALLOWED_HTML_TAGS).strip() c = Compound.create(current_package, compound_smiles, compound_name, compound_description) @@ -1193,9 +1198,10 @@ def package_compound(request, package_uuid, compound_uuid): return JsonResponse({"error": str(e)}, status=400) return JsonResponse({"success": current_compound.url}) - - new_compound_name = request.POST.get("compound-name", "").strip() - new_compound_description = request.POST.get("compound-description", "").strip() + # Clean for potential XSS + new_compound_name = nh3.clean(request.POST.get("compound-name", ""), tags=s.ALLOWED_HTML_TAGS).strip() + new_compound_description = nh3.clean(request.POST.get("compound-description", ""), + tags=s.ALLOWED_HTML_TAGS).strip() if new_compound_name: current_compound.name = new_compound_name @@ -1259,9 +1265,10 @@ def package_compound_structures(request, package_uuid, compound_uuid): return render(request, "collections/objects_list.html", context) elif request.method == "POST": - structure_name = request.POST.get("structure-name") - structure_smiles = request.POST.get("structure-smiles") - structure_description = request.POST.get("structure-description") + # Clean for potential XSS + structure_name = nh3.clean(request.POST.get("structure-name"), tags=s.ALLOWED_HTML_TAGS).strip() + structure_smiles = nh3.clean(request.POST.get("structure-smiles"), tags=s.ALLOWED_HTML_TAGS).strip() + structure_description = nh3.clean(request.POST.get("structure-description"), tags=s.ALLOWED_HTML_TAGS).strip() cs = current_compound.add_structure(structure_smiles, structure_name, structure_description) @@ -1321,9 +1328,11 @@ def package_compound_structure(request, package_uuid, compound_uuid, structure_u return redirect(current_compound.url + "/structure") else: return HttpResponseBadRequest() - - new_structure_name = request.POST.get("compound-structure-name", "").strip() - new_structure_description = request.POST.get("compound-structure-description", "").strip() + # Clean for potential XSS + new_structure_name = nh3.clean(request.POST.get("compound-structure-name", ""), + tags=s.ALLOWED_HTML_TAGS).strip() + new_structure_description = nh3.clean(request.POST.get("compound-structure-description", ""), + tags=s.ALLOWED_HTML_TAGS).strip() if new_structure_name: current_structure.name = new_structure_name @@ -1416,8 +1425,9 @@ def package_rules(request, package_uuid): log_post_params(request) # Generic params - rule_name = request.POST.get("rule-name") - rule_description = request.POST.get("rule-description") + # Clean for potential XSS + rule_name = nh3.clean(request.POST.get("rule-name"), tags=s.ALLOWED_HTML_TAGS).strip() + rule_description = nh3.clean(request.POST.get("rule-description"), tags=s.ALLOWED_HTML_TAGS).strip() rule_type = request.POST.get("rule-type") @@ -1425,11 +1435,16 @@ def package_rules(request, package_uuid): # Obtain parameters as required by rule type if rule_type == "SimpleAmbitRule": - params["smirks"] = request.POST.get("rule-smirks") - params["reactant_filter_smarts"] = request.POST.get("rule-reactant-smarts") - params["product_filter_smarts"] = request.POST.get("rule-product-smarts") + # Clean for potential XSS + params["smirks"] = nh3.clean(request.POST.get("rule-smirks"), tags=s.ALLOWED_HTML_TAGS).strip() + params["reactant_filter_smarts"] = nh3.clean(request.POST.get("rule-reactant-smarts"), + tags=s.ALLOWED_HTML_TAGS).strip() + params["product_filter_smarts"] = nh3.clean(request.POST.get("rule-product-smarts"), + tags=s.ALLOWED_HTML_TAGS).strip() elif rule_type == "SimpleRDKitRule": - params["reaction_smarts"] = request.POST.get("rule-reaction-smarts") + # Clean for potential XSS + params["reaction_smarts"] = nh3.clean(request.POST.get("rule-reaction-smarts"), + tags=s.ALLOWED_HTML_TAGS).strip() elif rule_type == "ParallelRule": pass elif rule_type == "SequentialRule": @@ -1522,8 +1537,9 @@ def package_rule(request, package_uuid, rule_uuid): return JsonResponse({"success": current_rule.url}) - rule_name = request.POST.get("rule-name", "").strip() - rule_description = request.POST.get("rule-description", "").strip() + # Clean for potential XSS + rule_name = nh3.clean(request.POST.get("rule-name", ""), tags=s.ALLOWED_HTML_TAGS).strip() + rule_description = nh3.clean(request.POST.get("rule-description", "").strip(), tags=s.ALLOWED_HTML_TAGS).strip() if rule_name: current_rule.name = rule_name @@ -1584,9 +1600,10 @@ def package_reactions(request, package_uuid): return render(request, "collections/objects_list.html", context) elif request.method == "POST": - reaction_name = request.POST.get("reaction-name") - reaction_description = request.POST.get("reaction-description") - reactions_smirks = request.POST.get("reaction-smirks") + # Clean for potential XSS + reaction_name = nh3.clean(request.POST.get("reaction-name"), tags=s.ALLOWED_HTML_TAGS).strip() + reaction_description = nh3.clean(request.POST.get("reaction-description"), tags=s.ALLOWED_HTML_TAGS).strip() + reactions_smirks = nh3.clean(request.POST.get("reaction-smirks"), tags=s.ALLOWED_HTML_TAGS).strip() educts = reactions_smirks.split(">>")[0].split(".") products = reactions_smirks.split(">>")[1].split(".") @@ -1648,8 +1665,10 @@ def package_reaction(request, package_uuid, reaction_uuid): return JsonResponse({"success": current_reaction.url}) - new_reaction_name = request.POST.get("reaction-name", "").strip() - new_reaction_description = request.POST.get("reaction-description", "").strip() + # Clean for potential XSS + new_reaction_name = nh3.clean(request.POST.get("reaction-name", ""), tags=s.ALLOWED_HTML_TAGS).strip() + new_reaction_description = nh3.clean(request.POST.get("reaction-description", ""), + tags=s.ALLOWED_HTML_TAGS).strip() if new_reaction_name: current_reaction.name = new_reaction_name @@ -1724,11 +1743,12 @@ def package_pathways(request, package_uuid): elif request.method == "POST": log_post_params(request) - name = request.POST.get("name") - # Sanitize for potential XSS - description = nh3.clean(request.POST.get("description"), tags=s.ALLOWED_HTML_TAGS) + # Clean for potential XSS + name = nh3.clean(request.POST.get("name"), tags=s.ALLOWED_HTML_TAGS).strip() + description = nh3.clean(request.POST.get("description"), tags=s.ALLOWED_HTML_TAGS).strip() + smiles = nh3.clean(request.POST.get("smiles", ""), tags=s.ALLOWED_HTML_TAGS).strip() + pw_mode = request.POST.get("predict", "predict").strip() - smiles = request.POST.get("smiles", "").strip() if "smiles" in request.POST and smiles == "": return error( @@ -1737,8 +1757,6 @@ def package_pathways(request, package_uuid): "Pathway prediction failed due to missing or empty SMILES", ) - smiles = smiles.strip() - try: stand_smiles = FormatConverter.standardize(smiles) except ValueError: @@ -1878,9 +1896,9 @@ def package_pathway(request, package_uuid, pathway_uuid): return JsonResponse({"success": current_pathway.url}) - pathway_name = request.POST.get("pathway-name") - # Sanitize for potential XSS - pathway_description = nh3.clean(request.POST.get("pathway-description"), tags=s.ALLOWED_HTML_TAGS) + # Clean for potential XSS + pathway_name = nh3.clean(request.POST.get("pathway-name"), tags=s.ALLOWED_HTML_TAGS).strip() + pathway_description = nh3.clean(request.POST.get("pathway-description"), tags=s.ALLOWED_HTML_TAGS).strip() if any([pathway_name, pathway_description]): if pathway_name is not None and pathway_name.strip() != "": @@ -1960,9 +1978,10 @@ def package_pathway_nodes(request, package_uuid, pathway_uuid): return render(request, "collections/objects_list.html", context) elif request.method == "POST": - node_name = request.POST.get("node-name") - node_description = request.POST.get("node-description") - node_smiles = request.POST.get("node-smiles") + # Clean for potential XSS + node_name = nh3.clean(request.POST.get("node-name"), tags=s.ALLOWED_HTML_TAGS).strip() + node_description = nh3.clean(request.POST.get("node-description"), tags=s.ALLOWED_HTML_TAGS).strip() + node_smiles = nh3.clean(request.POST.get("node-smiles"), tags=s.ALLOWED_HTML_TAGS).strip() current_pathway.add_node(node_smiles, name=node_name, description=node_description) @@ -2092,9 +2111,10 @@ def package_pathway_edges(request, package_uuid, pathway_uuid): elif request.method == "POST": log_post_params(request) + # Clean for potential XSS + edge_name = nh3.clean(request.POST.get("edge-name"), tags=s.ALLOWED_HTML_TAGS).strip() + edge_description = nh3.clean(request.POST.get("edge-description"), tags=s.ALLOWED_HTML_TAGS).strip() - edge_name = request.POST.get("edge-name") - edge_description = request.POST.get("edge-description") edge_substrates = request.POST.getlist("edge-substrates") edge_products = request.POST.getlist("edge-products") @@ -2181,7 +2201,7 @@ def package_scenarios(request, package_uuid): "all", False ): scens = Scenario.objects.filter(package=current_package).order_by("name") - res = [{"name": s.name, "url": s.url, "uuid": s.uuid} for s in scens] + res = [{"name": s_.name, "url": s_.url, "uuid": s_.uuid} for s_ in scens] return JsonResponse(res, safe=False) context = get_base_context(request) @@ -2229,21 +2249,21 @@ def package_scenarios(request, package_uuid): "name": "soil", "widgets": [ HTMLGenerator.generate_html(ai, prefix=f"soil_{0}") - for ai in [x for s in SOIL_ADDITIONAL_INFORMATION.values() for x in s] + for ai in [x for sv in SOIL_ADDITIONAL_INFORMATION.values() for x in sv] ], }, "Sludge Data": { "name": "sludge", "widgets": [ HTMLGenerator.generate_html(ai, prefix=f"sludge_{0}") - for ai in [x for s in SLUDGE_ADDITIONAL_INFORMATION.values() for x in s] + for ai in [x for sv in SLUDGE_ADDITIONAL_INFORMATION.values() for x in sv] ], }, "Water-Sediment System Data": { "name": "sediment", "widgets": [ HTMLGenerator.generate_html(ai, prefix=f"sediment_{0}") - for ai in [x for s in SEDIMENT_ADDITIONAL_INFORMATION.values() for x in s] + for ai in [x for sv in SEDIMENT_ADDITIONAL_INFORMATION.values() for x in sv] ], }, } @@ -2256,8 +2276,10 @@ def package_scenarios(request, package_uuid): elif request.method == "POST": log_post_params(request) - scenario_name = request.POST.get("scenario-name") - scenario_description = request.POST.get("scenario-description") + # Clean for potential XSS + scenario_name = nh3.clean(request.POST.get("scenario-name"), tags=s.ALLOWED_HTML_TAGS).strip() + scenario_description = nh3.clean(request.POST.get("scenario-description"), tags=s.ALLOWED_HTML_TAGS).strip() + scenario_date_year = request.POST.get("scenario-date-year") scenario_date_month = request.POST.get("scenario-date-month") scenario_date_day = request.POST.get("scenario-date-day") @@ -2270,10 +2292,10 @@ def package_scenarios(request, package_uuid): scenario_type = request.POST.get("scenario-type") - additional_information = HTMLGenerator.build_models(request.POST.dict()) - additional_information = [x for s in additional_information.values() for x in s] + additional_information = HTMLGenerator.build_models(clean_dict(request.POST.dict())) + additional_information = [x for sv in additional_information.values() for x in sv] - s = Scenario.create( + new_scen = Scenario.create( current_package, name=scenario_name, description=scenario_description, @@ -2282,7 +2304,7 @@ def package_scenarios(request, package_uuid): additional_information=additional_information, ) - return redirect(s.url) + return redirect(new_scen.url) else: return HttpResponseNotAllowed( [ @@ -2341,7 +2363,7 @@ def package_scenario(request, package_uuid, scenario_uuid): current_scenario.save() return redirect(current_scenario.url) elif hidden == "set-additional-information": - post_dict = sanitize_dict(request.POST.dict()) # Sanitise post dict inputs for potential XSS + post_dict = clean_dict(request.POST.dict()) # Clean post dict inputs for potential XSS ais = HTMLGenerator.build_models(post_dict) if s.DEBUG: @@ -2350,7 +2372,7 @@ def package_scenario(request, package_uuid, scenario_uuid): current_scenario.set_additional_information(ais) return redirect(current_scenario.url) elif hidden == "add-additional-information": - post_dict = sanitize_dict(request.POST.dict()) # Sanitise post dict inputs for potential XSS + post_dict = clean_dict(request.POST.dict()) # Clean post dict inputs for potential XSS ais = HTMLGenerator.build_models(post_dict) if len(ais.keys()) != 1: @@ -2491,8 +2513,10 @@ def groups(request): return render(request, "collections/objects_list.html", context) elif request.method == "POST": - group_name = request.POST.get("group-name") - group_description = request.POST.get("group-description", s.DEFAULT_VALUES["description"]) + # Clean for potential XSS + group_name = nh3.clean(request.POST.get("group-name"), tags=s.ALLOWED_HTML_TAGS).strip() + group_description = nh3.clean(request.POST.get("group-description", s.DEFAULT_VALUES["description"]), + tags=s.ALLOWED_HTML_TAGS).strip() g = GroupManager.create_group(current_user, group_name, group_description) @@ -2582,8 +2606,10 @@ def settings(request): logger.info("Parameters received:") logger.info(f"{k}\t{v}") - name = request.POST.get("prediction-setting-name") - description = request.POST.get("prediction-setting-description") + # Clean for potential XSS + name = nh3.clean(request.POST.get("prediction-setting-name"), tags=s.ALLOWED_HTML_TAGS).strip() + description = nh3.clean(request.POST.get("prediction-setting-description"), tags=s.ALLOWED_HTML_TAGS).strip() + new_default = request.POST.get("prediction-setting-new-default", "off") == "on" max_nodes = min(