resource-screener

Name: resource-screener
Author: KnoxOps

38stars4forks

How to install resource-screener?

knoxops/open-devops-skills/resource-screener

$npx -y skills add KnoxOps/open-devops-skills --skill resource-screener

Installs into the current project.

Summary

Multi-dimensional zombie resource scoring and priority ranking - 4 simplified unified scoring dimensions (cpu_idle, network_idle, ownership_clarity, data_sensitivity) - Environment-adjusted suspect_level thresholds - One-vote veto protection rules - Priority formula: zombie_score

Files · 1

View on GitHub

SKILL.md

1	## Input Parameters
2
3	\| Name \| Type \| Required \| Description \|
4	\|------\|------\|----------\|-------------\|
5	\| run_dir \| string \| Yes \| Work directory absolute path \|
6	\| ssh_key_path \| string \| No \| SSH key path for SSH verification and cloud tags queries \|
7	\| task_id \| string \| Yes \| Task ID for progress tracking \|
8
9	## Execution Flow
10
11	### Task Context
12
13	Before starting execution, initialize `task_context.json`:
14
15	```json
16	{
17	"task_id": "<task_id from input>",
18	"current_step": 0,
19	"current_step_id": null,
20	"status": "running",
21	"steps": {
22	"setup-layer-d-skeleton": "pending",
23	"execute-scoring-pipeline": "pending",
24	"generate-per-resource-files": "pending",
25	"generate-suspect-assessment-json": "pending",
26	"write-output-files": "pending"
27	},
28	"updated_at": "<ISO timestamp>"
29	}
30	```
31
32	Update this file after each step completes. On error, set step status to `"failed"` and overall `status` to `"failed"`.
33
34	### Step 1: setup-layer-d-skeleton
35
36	Type: inline
37	Description: Input validation and initialization
38
39	## Execution
40	Follow these instructions:
41
42	import os
43	import json
44	from datetime import datetime
45
46	def run(run_dir, task_id):
47	# Verify input files exist
48	files_to_check = [
49	"layer_b_candidates.json",
50	"scan_plan.json"
51	]
52
53	missing = []
54	for f in files_to_check:
55	path = os.path.join(run_dir, f)
56	if not os.path.exists(path):
57	missing.append(f)
58
59	if missing:
60	return {"status": "failed", "errors": f"Missing files: {missing}"}
61
62	# Read and validate key fields
63	try:
64	with open(os.path.join(run_dir, "layer_b_candidates.json")) as f:
65	candidates_data = json.load(f)
66	if "candidates" not in candidates_data:
67	return {"status": "failed", "error": "layer_b_candidates.json missing 'candidates' array"}
68	if len(candidates_data["candidates"]) == 0:
69	return {"status": "empty", "message": "No candidates after Layer B filtering"}
70
71	# Validate collector signal fields exist on first candidate
72	first = candidates_data["candidates"][0]
73	required_signals = ["resource_id", "resource_type", "entity_type"]
74	missing_signals = [s for s in required_signals if s not in first]
75	if missing_signals:
76	return {"status": "failed", "error": f"candidates missing required fields: {missing_signals}"}
77
78	# Optional signal fields (collector may provide these)
79	# cpu_avg_pct, network_mb_per_day, has_human_login, has_real_alert
80	# owner_status, dependencies, reachability
81
82	with open(os.path.join(run_dir, "scan_plan.json")) as f:
83	scan_plan = json.load(f)
84	if "dimension_framework" not in scan_plan:
85	return {"status": "failed", "error": "scan_plan.json missing 'dimension_framework'"}
86	except json.JSONDecodeError as e:
87	return {"status": "failed", "error": f"JSON parse error: {e}"}
88
89	# Create output directory
90	os.makedirs(os.path.join(run_dir, "analysis"), exist_ok=True)
91
92	return {
93	"status": "success",
94	"candidates_count": len(candidates_data["candidates"]),
95	"prepared_at": datetime.utcnow().isoformat() + "Z"
96	}
97
98
99	### Progress Tracking
100
101	After completing this step, update `task_context.json`:
102	- Set `current_step_id` to `"setup-layer-d-skeleton"`
103	- Set `steps.setup-layer-d-skeleton` to `"completed"`
104	### Step 2: execute-scoring-pipeline
105
106	Type: inline
107	Description: Execute full scoring pipeline on all candidates
108
109	## Execution
110	Follow these instructions:
111
112	import json
113	import os
114	import math
115
116	# Unified dimension weights (same for all resource types)
117	# Focus: CPU idleness, network idleness, ownership clarity, data sensitivity
118	# Heavier score on cpu+network near-zero = higher zombie suspicion
119	DIMENSION_WEIGHTS = {
120	"cpu_idle": 0.35,
121	"network_idle": 0.35,
122	"ownership_clarity": 0.20,
123	"data_sensitivity": 0.10
124	}
125
126	# === Signal → Dimension Score Converters ===
127
128	def compute_cpu_idle_score(cpu_avg_pct):
129	"""Lower CPU peak = more idle = higher zombie suspicion.
130	Returns (score, reliability)."""
131	if cpu_avg_pct is None:
132	return 0.0, 0.0
133	if cpu_avg_pct <= 1.0:
134	return 0.95, 1.0
135	elif cpu_avg_pct <= 3.0:
136	return 0.80, 1.0
137	elif cpu_avg_pct <= 5.0:
138	return 0.50, 1.0
139	elif cpu_avg_pct <= 10.0:
140	return 0.20, 1.0
141	else:
142	return 0.0, 1.0
143
144	def compute_network_idle_score(network_mb_per_day):
145	"""Lower network throughput = more idle = higher zombie suspicion.
146	Returns (score, reliability)."""
147	if network_mb_per_day is None:
148	return 0.0, 0.0
149	if network_mb_per_day <= 0.1:
150	return 0.95, 1.0
151	elif network_mb_per_day <= 1.0:
152	return 0.80, 1.0
153	elif network_mb_per_da

People who liked this also viewed

6 picks

Type

resource-screener

38stars4forks

View on GitHub

How to install resource-screener?

knoxops/open-devops-skills/resource-screener

$npx -y skills add KnoxOps/open-devops-skills --skill resource-screener

Installs into the current project.

Summary

Files · 1

View on GitHub

SKILL.md

1	## Input Parameters
2
3	\| Name \| Type \| Required \| Description \|
4	\|------\|------\|----------\|-------------\|
5	\| run_dir \| string \| Yes \| Work directory absolute path \|
6	\| ssh_key_path \| string \| No \| SSH key path for SSH verification and cloud tags queries \|
7	\| task_id \| string \| Yes \| Task ID for progress tracking \|
8
9	## Execution Flow
10
11	### Task Context
12
13	Before starting execution, initialize `task_context.json`:
14
15	```json
16	{
17	"task_id": "<task_id from input>",
18	"current_step": 0,
19	"current_step_id": null,
20	"status": "running",
21	"steps": {
22	"setup-layer-d-skeleton": "pending",
23	"execute-scoring-pipeline": "pending",
24	"generate-per-resource-files": "pending",
25	"generate-suspect-assessment-json": "pending",
26	"write-output-files": "pending"
27	},
28	"updated_at": "<ISO timestamp>"
29	}
30	```
31
32	Update this file after each step completes. On error, set step status to `"failed"` and overall `status` to `"failed"`.
33
34	### Step 1: setup-layer-d-skeleton
35
36	Type: inline
37	Description: Input validation and initialization
38
39	## Execution
40	Follow these instructions:
41
42	import os
43	import json
44	from datetime import datetime
45
46	def run(run_dir, task_id):
47	# Verify input files exist
48	files_to_check = [
49	"layer_b_candidates.json",
50	"scan_plan.json"
51	]
52
53	missing = []
54	for f in files_to_check:
55	path = os.path.join(run_dir, f)
56	if not os.path.exists(path):
57	missing.append(f)
58
59	if missing:
60	return {"status": "failed", "errors": f"Missing files: {missing}"}
61
62	# Read and validate key fields
63	try:
64	with open(os.path.join(run_dir, "layer_b_candidates.json")) as f:
65	candidates_data = json.load(f)
66	if "candidates" not in candidates_data:
67	return {"status": "failed", "error": "layer_b_candidates.json missing 'candidates' array"}
68	if len(candidates_data["candidates"]) == 0:
69	return {"status": "empty", "message": "No candidates after Layer B filtering"}
70
71	# Validate collector signal fields exist on first candidate
72	first = candidates_data["candidates"][0]
73	required_signals = ["resource_id", "resource_type", "entity_type"]
74	missing_signals = [s for s in required_signals if s not in first]
75	if missing_signals:
76	return {"status": "failed", "error": f"candidates missing required fields: {missing_signals}"}
77
78	# Optional signal fields (collector may provide these)
79	# cpu_avg_pct, network_mb_per_day, has_human_login, has_real_alert
80	# owner_status, dependencies, reachability
81
82	with open(os.path.join(run_dir, "scan_plan.json")) as f:
83	scan_plan = json.load(f)
84	if "dimension_framework" not in scan_plan:
85	return {"status": "failed", "error": "scan_plan.json missing 'dimension_framework'"}
86	except json.JSONDecodeError as e:
87	return {"status": "failed", "error": f"JSON parse error: {e}"}
88
89	# Create output directory
90	os.makedirs(os.path.join(run_dir, "analysis"), exist_ok=True)
91
92	return {
93	"status": "success",
94	"candidates_count": len(candidates_data["candidates"]),
95	"prepared_at": datetime.utcnow().isoformat() + "Z"
96	}
97
98
99	### Progress Tracking
100
101	After completing this step, update `task_context.json`:
102	- Set `current_step_id` to `"setup-layer-d-skeleton"`
103	- Set `steps.setup-layer-d-skeleton` to `"completed"`
104	### Step 2: execute-scoring-pipeline
105
106	Type: inline
107	Description: Execute full scoring pipeline on all candidates
108
109	## Execution
110	Follow these instructions:
111
112	import json
113	import os
114	import math
115
116	# Unified dimension weights (same for all resource types)
117	# Focus: CPU idleness, network idleness, ownership clarity, data sensitivity
118	# Heavier score on cpu+network near-zero = higher zombie suspicion
119	DIMENSION_WEIGHTS = {
120	"cpu_idle": 0.35,
121	"network_idle": 0.35,
122	"ownership_clarity": 0.20,
123	"data_sensitivity": 0.10
124	}
125
126	# === Signal → Dimension Score Converters ===
127
128	def compute_cpu_idle_score(cpu_avg_pct):
129	"""Lower CPU peak = more idle = higher zombie suspicion.
130	Returns (score, reliability)."""
131	if cpu_avg_pct is None:
132	return 0.0, 0.0
133	if cpu_avg_pct <= 1.0:
134	return 0.95, 1.0
135	elif cpu_avg_pct <= 3.0:
136	return 0.80, 1.0
137	elif cpu_avg_pct <= 5.0:
138	return 0.50, 1.0
139	elif cpu_avg_pct <= 10.0:
140	return 0.20, 1.0
141	else:
142	return 0.0, 1.0
143
144	def compute_network_idle_score(network_mb_per_day):
145	"""Lower network throughput = more idle = higher zombie suspicion.
146	Returns (score, reliability)."""
147	if network_mb_per_day is None:
148	return 0.0, 0.0
149	if network_mb_per_day <= 0.1:
150	return 0.95, 1.0
151	elif network_mb_per_day <= 1.0:
152	return 0.80, 1.0
153	elif network_mb_per_da

People who liked this also viewed

6 picks

Type