|
18 | 18 | statement TEXT, |
19 | 19 | guidance TEXT, |
20 | 20 | baselines TEXT, |
21 | | - is_withdrawn INTEGER DEFAULT 0 |
| 21 | + is_withdrawn INTEGER DEFAULT 0, |
| 22 | + assessment_objectives TEXT, |
| 23 | + assessment_methods TEXT |
22 | 24 | ); |
23 | 25 | CREATE INDEX IF NOT EXISTS idx_nist_ctrl_family ON nist_controls(family_id); |
| 26 | +CREATE INDEX IF NOT EXISTS idx_nist_ctrl_baselines ON nist_controls(baselines); |
24 | 27 | """ |
25 | 28 |
|
26 | 29 | FTS_SQL = """ |
27 | 30 | CREATE VIRTUAL TABLE IF NOT EXISTS nist_controls_fts USING fts5( |
28 | | - id, family_name, title, statement, guidance, |
| 31 | + id, family_name, title, statement, guidance, assessment_objectives, |
29 | 32 | content='nist_controls', content_rowid='rowid' |
30 | 33 | ); |
31 | 34 | """ |
@@ -54,62 +57,55 @@ def _extract_prose(parts: list[dict[str, Any]] | None) -> str: |
54 | 57 | return " ".join(texts) |
55 | 58 |
|
56 | 59 |
|
57 | | -def _parse_controls(groups: list[dict], baseline_ids: set[str]) -> list[tuple]: |
| 60 | +def _extract_assessment(parts: list[dict[str, Any]]) -> tuple[str, str]: |
| 61 | + objectives = _extract_prose([p for p in parts if p.get("name") == "assessment-objective"]) |
| 62 | + methods_parts = [p for p in parts if p.get("name") == "assessment-method"] |
| 63 | + methods = [] |
| 64 | + for mp in methods_parts: |
| 65 | + method_id = mp.get("id", "") |
| 66 | + method_type = "examine" if "examine" in method_id else "interview" if "interview" in method_id else "test" |
| 67 | + objects_prose = _extract_prose([sp for sp in mp.get("parts", []) if sp.get("name") == "assessment-objects"]) |
| 68 | + if objects_prose: |
| 69 | + methods.append(f"[{method_type.upper()}] {objects_prose[:500]}") |
| 70 | + return objectives[:3000], "; ".join(methods)[:3000] |
| 71 | + |
| 72 | + |
| 73 | +def _parse_one_control(ctrl: dict, family_id: str, family_name: str, baseline_ids: dict) -> tuple: |
| 74 | + ctrl_id = ctrl.get("id", "") |
| 75 | + title = ctrl.get("title", "") |
| 76 | + is_withdrawn = 1 if any( |
| 77 | + True for p in ctrl.get("props", []) |
| 78 | + if p.get("name") == "status" and p.get("value") == "withdrawn" |
| 79 | + ) else 0 |
| 80 | + |
| 81 | + parts = ctrl.get("parts", []) |
| 82 | + statement = _extract_prose([p for p in parts if p.get("name") == "statement"]) |
| 83 | + guidance = _extract_prose([p for p in parts if p.get("name") == "guidance"]) |
| 84 | + assess_obj, assess_methods = _extract_assessment(parts) |
| 85 | + |
| 86 | + baselines = [] |
| 87 | + norm_id = ctrl_id.lower() |
| 88 | + for level in ["LOW", "MODERATE", "HIGH"]: |
| 89 | + if norm_id in baseline_ids.get(level, set()): |
| 90 | + baselines.append(level) |
| 91 | + |
| 92 | + return ( |
| 93 | + ctrl_id, family_id, family_name, title, |
| 94 | + statement[:5000], guidance[:5000], |
| 95 | + ",".join(baselines), is_withdrawn, |
| 96 | + assess_obj, assess_methods, |
| 97 | + ) |
| 98 | + |
| 99 | + |
| 100 | +def _parse_controls(groups: list[dict], baseline_ids: dict) -> list[tuple]: |
58 | 101 | rows = [] |
59 | 102 | for group in groups: |
60 | 103 | family_id = group.get("id", "") |
61 | 104 | family_name = group.get("title", "") |
62 | 105 | for ctrl in group.get("controls", []): |
63 | | - ctrl_id = ctrl.get("id", "") |
64 | | - title = ctrl.get("title", "") |
65 | | - is_withdrawn = 1 if any( |
66 | | - p.get("class") == "assessment" for p in ctrl.get("props", []) |
67 | | - if p.get("name") == "status" and p.get("value") == "withdrawn" |
68 | | - ) else 0 |
69 | | - |
70 | | - statement = _extract_prose( |
71 | | - [p for p in ctrl.get("parts", []) if p.get("name") == "statement"] |
72 | | - ) |
73 | | - guidance = _extract_prose( |
74 | | - [p for p in ctrl.get("parts", []) if p.get("name") == "guidance"] |
75 | | - ) |
76 | | - |
77 | | - baselines = [] |
78 | | - norm_id = ctrl_id.lower() |
79 | | - for level in ["LOW", "MODERATE", "HIGH"]: |
80 | | - if norm_id in baseline_ids.get(level, set()): |
81 | | - baselines.append(level) |
82 | | - |
83 | | - rows.append(( |
84 | | - ctrl_id, family_id, family_name, title, |
85 | | - statement[:5000] if statement else "", |
86 | | - guidance[:5000] if guidance else "", |
87 | | - ",".join(baselines) if baselines else "", |
88 | | - is_withdrawn, |
89 | | - )) |
90 | | - |
| 106 | + rows.append(_parse_one_control(ctrl, family_id, family_name, baseline_ids)) |
91 | 107 | for enh in ctrl.get("controls", []): |
92 | | - enh_id = enh.get("id", "") |
93 | | - enh_title = enh.get("title", "") |
94 | | - enh_statement = _extract_prose( |
95 | | - [p for p in enh.get("parts", []) if p.get("name") == "statement"] |
96 | | - ) |
97 | | - enh_guidance = _extract_prose( |
98 | | - [p for p in enh.get("parts", []) if p.get("name") == "guidance"] |
99 | | - ) |
100 | | - enh_baselines = [] |
101 | | - enh_norm = enh_id.lower() |
102 | | - for level in ["LOW", "MODERATE", "HIGH"]: |
103 | | - if enh_norm in baseline_ids.get(level, set()): |
104 | | - enh_baselines.append(level) |
105 | | - |
106 | | - rows.append(( |
107 | | - enh_id, family_id, family_name, enh_title, |
108 | | - enh_statement[:5000] if enh_statement else "", |
109 | | - enh_guidance[:5000] if enh_guidance else "", |
110 | | - ",".join(enh_baselines) if enh_baselines else "", |
111 | | - 0, |
112 | | - )) |
| 108 | + rows.append(_parse_one_control(enh, family_id, family_name, baseline_ids)) |
113 | 109 | return rows |
114 | 110 |
|
115 | 111 |
|
@@ -142,8 +138,8 @@ def scrape_nist_controls(conn: sqlite3.Connection) -> int: |
142 | 138 |
|
143 | 139 | conn.executemany( |
144 | 140 | "INSERT OR REPLACE INTO nist_controls " |
145 | | - "(id, family_id, family_name, title, statement, guidance, baselines, is_withdrawn) " |
146 | | - "VALUES (?, ?, ?, ?, ?, ?, ?, ?)", |
| 141 | + "(id, family_id, family_name, title, statement, guidance, baselines, is_withdrawn, assessment_objectives, assessment_methods) " |
| 142 | + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", |
147 | 143 | rows, |
148 | 144 | ) |
149 | 145 | conn.commit() |
|
0 commit comments