{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Check detailed samples\n",
"merge and check"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"import json\n",
"from utils.sample_utils import Sample\n",
"import os\n",
"from utils.file_utils import dir_check\n",
"checked_dir = \"data/checked\"\n",
"dir_check(checked_dir)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## check cve samples"
]
},
{
"cell_type": "code",
"execution_count": 2,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Filtered cve sample num: 99\n",
"{'NO_SAFETY_CONSTRAINT': 38, 'NO_UNSAFE_CALLEE': 28}\n"
]
}
],
"source": [
"cve_dir = \"data/detailed_cve_func\"\n",
"checked_cve_file = os.path.join(checked_dir, \"cve.json\")\n",
"checked_cve_samples = list()\n",
"cve_error_cnt = dict()\n",
"\n",
"for file in os.listdir(cve_dir):\n",
" cve_samples = json.load(open(os.path.join(cve_dir, file)))\n",
" for raw in cve_samples:\n",
" sample = Sample(raw)\n",
" if not sample.has_unsafe_callee():\n",
" cve_error_cnt[\"NO_UNSAFE_CALLEE\"] = (\n",
" cve_error_cnt.get(\"NO_UNSAFE_CALLEE\", 0) + 1\n",
" )\n",
" elif not sample.has_safety_constraint():\n",
" cve_error_cnt[\"NO_SAFETY_CONSTRAINT\"] = (\n",
" cve_error_cnt.get(\"NO_SAFETY_CONSTRAINT\", 0) + 1\n",
" )\n",
" else:\n",
" # add tags field to sample for further manually checking\n",
" raw[\"tags\"] = []\n",
" checked_cve_samples.append(raw)\n",
"print(\"Filtered cve sample num: \", len(checked_cve_samples))\n",
"json.dump(checked_cve_samples, open(checked_cve_file, \"w\"), indent=2)\n",
"print(cve_error_cnt)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## check unsafe samples"
]
},
{
"cell_type": "code",
"execution_count": 2,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Filtered unsafe sample num: 144\n",
"{'NO_OTHER_PARAMETER': 33, 'NO_SAFETY_CONSTRAINT': 197, 'NO_UNSAFE_CALLEE': 990, 'TOO_LONG_CODE': 8, 'NO_SAFETY_COMMENT': 62}\n"
]
}
],
"source": [
"unsafe_dir = \"data/detailed_unsafe_func\"\n",
"checked_unsafe_file = os.path.join(checked_dir, \"unsafe.json\")\n",
"checked_unsafe_samples = list()\n",
"unsafe_error_cnt = dict()\n",
"\n",
"for file in os.listdir(unsafe_dir):\n",
" unsafe_samples = json.load(open(os.path.join(unsafe_dir, file)))\n",
" for raw in unsafe_samples:\n",
" sample = Sample(raw)\n",
" if not sample.has_unsafe_callee():\n",
" unsafe_error_cnt[\"NO_UNSAFE_CALLEE\"] = (\n",
" unsafe_error_cnt.get(\"NO_UNSAFE_CALLEE\", 0) + 1\n",
" )\n",
" elif not sample.has_safety_constraint():\n",
" unsafe_error_cnt[\"NO_SAFETY_CONSTRAINT\"] = (\n",
" unsafe_error_cnt.get(\"NO_SAFETY_CONSTRAINT\", 0) + 1\n",
" )\n",
" elif \"Safety\" not in sample.fn_comment and \"SAFETY\" not in sample.fn_comment:\n",
" unsafe_error_cnt[\"NO_SAFETY_COMMENT\"] = (\n",
" unsafe_error_cnt.get(\"NO_SAFETY_COMMENT\", 0) + 1\n",
" )\n",
" elif not sample.has_other_parameter():\n",
" unsafe_error_cnt[\"NO_OTHER_PARAMETER\"] = (\n",
" unsafe_error_cnt.get(\"NO_OTHER_PARAMETER\", 0) + 1\n",
" )\n",
" elif sample.nested_fn():\n",
" unsafe_error_cnt[\"NESTED_FN\"] = (\n",
" unsafe_error_cnt.get(\"NESTED_FN\", 0) + 1\n",
" )\n",
" elif sample.too_long_code():\n",
" unsafe_error_cnt[\"TOO_LONG_CODE\"] = (\n",
" unsafe_error_cnt.get(\"TOO_LONG_CODE\", 0) + 1\n",
" )\n",
" else:\n",
" checked_unsafe_samples.append(raw)\n",
"print(\"Filtered unsafe sample num: \", len(checked_unsafe_samples))\n",
"json.dump(checked_unsafe_samples, open(checked_unsafe_file, \"w\"), indent=2)\n",
"print(unsafe_error_cnt)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## check risky samples"
]
},
{
"cell_type": "code",
"execution_count": 3,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Filtered risky sample num: 244\n",
"{'NO_UNSAFE_CALLEE': 825, 'NO_SAFETY_COMMENT': 389, 'NO_SAFETY_CONSTRAINT': 458, 'NO_OTHER_PARAMETER': 94, 'NESTED_FN': 2}\n"
]
}
],
"source": [
"risky_dir = \"data/detailed_risky_func\"\n",
"checked_risky_file = os.path.join(checked_dir, \"risky.json\")\n",
"checked_risky_samples = list()\n",
"risky_error_cnt = dict()\n",
"\n",
"for file in os.listdir(risky_dir):\n",
" risky_samples = json.load(open(os.path.join(risky_dir, file)))\n",
" for raw in risky_samples:\n",
" sample = Sample(raw)\n",
" if not sample.has_unsafe_callee():\n",
" risky_error_cnt[\"NO_UNSAFE_CALLEE\"] = (\n",
" risky_error_cnt.get(\"NO_UNSAFE_CALLEE\", 0) + 1\n",
" )\n",
" elif not sample.has_other_parameter():\n",
" risky_error_cnt[\"NO_OTHER_PARAMETER\"] = (\n",
" risky_error_cnt.get(\"NO_OTHER_PARAMETER\", 0) + 1\n",
" )\n",
" elif sample.nested_fn():\n",
" risky_error_cnt[\"NESTED_FN\"] = (\n",
" risky_error_cnt.get(\"NESTED_FN\", 0) + 1\n",
" )\n",
" elif sample.too_long_code():\n",
" risky_error_cnt[\"TOO_LONG_CODE\"] = (\n",
" risky_error_cnt.get(\"TOO_LONG_CODE\", 0) + 1\n",
" )\n",
" elif not sample.has_safety_constraint():\n",
" risky_error_cnt[\"NO_SAFETY_CONSTRAINT\"] = (\n",
" risky_error_cnt.get(\"NO_SAFETY_CONSTRAINT\", 0) + 1\n",
" )\n",
" elif not sample.has_safety_comment():\n",
" risky_error_cnt[\"NO_SAFETY_COMMENT\"] = (\n",
" risky_error_cnt.get(\"NO_SAFETY_COMMENT\", 0) + 1\n",
" )\n",
" else:\n",
" checked_risky_samples.append(raw)\n",
"print(\"Filtered risky sample num: \", len(checked_risky_samples))\n",
"json.dump(checked_risky_samples, open(checked_risky_file, \"w\"), indent=2)\n",
"print(risky_error_cnt)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Check Scan Samples"
]
},
{
"cell_type": "code",
"execution_count": 3,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Filtered risky sample num: 552\n",
"{'NO_UNSAFE_CALLEE': 1121, 'NO_SAFETY_CONSTRAINT': 752, 'NO_OTHER_PARAMETER': 319, 'NESTED_FN': 21, 'TOO_LONG_CODE': 23}\n"
]
}
],
"source": [
"def check_samples_in_target(target):\n",
" checked_scan_samples = list()\n",
" scan_error_cnt = dict()\n",
" scan_dir = f\"data/detailed_{target}_func\"\n",
" checked_scan_file = os.path.join(checked_dir, \"scan.json\")\n",
" for file in os.listdir(scan_dir):\n",
" risky_samples = json.load(open(os.path.join(risky_dir, file)))\n",
" for raw in risky_samples:\n",
" sample = Sample(raw)\n",
" if not sample.has_unsafe_callee():\n",
" scan_error_cnt[\"NO_UNSAFE_CALLEE\"] = (\n",
" scan_error_cnt.get(\"NO_UNSAFE_CALLEE\", 0) + 1\n",
" )\n",
" elif not sample.has_other_parameter():\n",
" scan_error_cnt[\"NO_OTHER_PARAMETER\"] = (\n",
" scan_error_cnt.get(\"NO_OTHER_PARAMETER\", 0) + 1\n",
" )\n",
" elif sample.nested_fn():\n",
" scan_error_cnt[\"NESTED_FN\"] = (\n",
" scan_error_cnt.get(\"NESTED_FN\", 0) + 1\n",
" )\n",
" elif sample.too_long_code():\n",
" scan_error_cnt[\"TOO_LONG_CODE\"] = (\n",
" scan_error_cnt.get(\"TOO_LONG_CODE\", 0) + 1\n",
" )\n",
" elif not sample.has_safety_constraint():\n",
" scan_error_cnt[\"NO_SAFETY_CONSTRAINT\"] = (\n",
" scan_error_cnt.get(\"NO_SAFETY_CONSTRAINT\", 0) + 1\n",
" )\n",
" # elif not sample.has_safety_comment():\n",
" # risky_error_cnt[\"NO_SAFETY_COMMENT\"] = (\n",
" # risky_error_cnt.get(\"NO_SAFETY_COMMENT\", 0) + 1\n",
" # )\n",
" else:\n",
" checked_scan_samples.append(raw)\n",
" print(f\"Filtered risky {len(checked_scan_samples)} samples in {target}\")\n",
" json.dump(checked_scan_samples, open(checked_scan_file, \"w\"), indent=2)\n",
" print(scan_error_cnt)\n",
"\n",
"check_samples_in_target(\"scan\")\n",
"check_samples_in_target(\"scan1k_2k\")\n",
"check_samples_in_target(\"scan10k_15k\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Preprocess Samples\n",
"- Document Rewrite: rewrite the documentation of unsafe APIs that reference to other unsafe API (written in hyperlink format);\n",
"- Remove Duplicated: remove samples that are involved in examples;"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"import json\n",
"import os\n",
"from utils.file_utils import dir_check\n",
"dir_check(\"data/samples/original\")\n",
"\n",
"example_file = \"examples/examples_for_all_guarantee_patterns.json\"\n",
"examples = json.load(open(example_file, \"r\"))\n",
"example_labels = [example[\"sample_label\"] for example in examples]\n",
"\n",
"rewrite_file = \"examples/document_rewrite.json\"\n",
"rewrites = json.load(open(rewrite_file, \"r\"))\n",
"fn_names = {r[\"name\"] for r in rewrites}\n",
"rewrite_rules = [(r[\"src\"], r[\"dst\"]) for r in rewrites]\n",
"\n",
"targets = [\"filtered_unsafe\", \"risky\", \"11cve\"]\n",
"for target in targets:\n",
" filtered_samples = []\n",
" file = os.path.join(\"data\", \"checked\", target + \".json\")\n",
" samples = json.load(open(file, \"r\"))\n",
" for sample in samples:\n",
" if sample[\"sample_label\"] not in example_labels:\n",
" continue\n",
" for info in sample[\"detail_info\"]:\n",
" if info[\"item_type\"] == \"function\" and info[\"name\"] in fn_names:\n",
" for rule in rewrite_rules:\n",
" sources, dst = rule\n",
" for src in sources:\n",
" info[\"documentation\"] = info[\"documentation\"].replace(src, dst)\n",
" filtered_samples.append(sample)\n",
" result_file = os.path.join(\"data\", \"samples\", \"original\", target + \".json\")\n",
" json.dump(filtered_samples, open(result_file, \"w\"), indent=2)"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "unsafe",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.10.13"
}
},
"nbformat": 4,
"nbformat_minor": 2
}