mirror of
https://github.com/smittix/intercept.git
synced 2026-07-11 19:18:13 -07:00
Remove Drone Ops feature end-to-end
This commit is contained in:
Smittix
@@ -1,36 +0,0 @@
|
||||
"""Tests for Drone Ops policy helpers and service policy behavior."""
|
||||
|
||||
from utils.drone.policy import required_approvals_for_action
|
||||
from utils.drone.service import DroneOpsService
|
||||
|
||||
|
||||
def test_required_approvals_policy_helper():
|
||||
assert required_approvals_for_action('passive_scan') == 1
|
||||
assert required_approvals_for_action('wifi_deauth_test') == 2
|
||||
|
||||
|
||||
def test_service_required_approvals_matches_policy_helper():
|
||||
assert DroneOpsService.required_approvals('passive_capture') == required_approvals_for_action('passive_capture')
|
||||
assert DroneOpsService.required_approvals('active_test') == required_approvals_for_action('active_test')
|
||||
|
||||
|
||||
def test_service_arm_disarm_policy_state():
|
||||
service = DroneOpsService()
|
||||
|
||||
armed = service.arm_actions(
|
||||
actor='operator-1',
|
||||
reason='controlled testing',
|
||||
incident_id=42,
|
||||
duration_seconds=5,
|
||||
)
|
||||
assert armed['armed'] is True
|
||||
assert armed['armed_by'] == 'operator-1'
|
||||
assert armed['arm_reason'] == 'controlled testing'
|
||||
assert armed['arm_incident_id'] == 42
|
||||
assert armed['armed_until'] is not None
|
||||
|
||||
disarmed = service.disarm_actions(actor='operator-1', reason='test complete')
|
||||
assert disarmed['armed'] is False
|
||||
assert disarmed['armed_by'] is None
|
||||
assert disarmed['arm_reason'] is None
|
||||
assert disarmed['arm_incident_id'] is None
|
||||
@@ -1,60 +0,0 @@
|
||||
"""Tests for Drone Ops Remote ID decoder helpers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
from utils.drone.remote_id import decode_remote_id_payload
|
||||
|
||||
|
||||
def test_decode_remote_id_from_dict_payload():
|
||||
payload = {
|
||||
'remote_id': {
|
||||
'uas_id': 'UAS-001',
|
||||
'operator_id': 'OP-007',
|
||||
'lat': 37.7749,
|
||||
'lon': -122.4194,
|
||||
'altitude_m': 121.5,
|
||||
'speed_mps': 12.3,
|
||||
'heading_deg': 270.0,
|
||||
}
|
||||
}
|
||||
|
||||
decoded = decode_remote_id_payload(payload)
|
||||
assert decoded['detected'] is True
|
||||
assert decoded['source_format'] == 'dict'
|
||||
assert decoded['uas_id'] == 'UAS-001'
|
||||
assert decoded['operator_id'] == 'OP-007'
|
||||
assert decoded['lat'] == 37.7749
|
||||
assert decoded['lon'] == -122.4194
|
||||
assert decoded['altitude_m'] == 121.5
|
||||
assert decoded['speed_mps'] == 12.3
|
||||
assert decoded['heading_deg'] == 270.0
|
||||
assert decoded['confidence'] > 0.0
|
||||
|
||||
|
||||
def test_decode_remote_id_from_json_string():
|
||||
payload = json.dumps({
|
||||
'uas_id': 'RID-ABC',
|
||||
'lat': 35.0,
|
||||
'lon': -115.0,
|
||||
'altitude': 80,
|
||||
})
|
||||
|
||||
decoded = decode_remote_id_payload(payload)
|
||||
assert decoded['detected'] is True
|
||||
assert decoded['source_format'] == 'json'
|
||||
assert decoded['uas_id'] == 'RID-ABC'
|
||||
assert decoded['lat'] == 35.0
|
||||
assert decoded['lon'] == -115.0
|
||||
assert decoded['altitude_m'] == 80.0
|
||||
|
||||
|
||||
def test_decode_remote_id_from_raw_text_is_not_detected():
|
||||
decoded = decode_remote_id_payload('not-a-remote-id-payload')
|
||||
assert decoded['detected'] is False
|
||||
assert decoded['source_format'] == 'raw'
|
||||
assert decoded['uas_id'] is None
|
||||
assert decoded['operator_id'] is None
|
||||
assert isinstance(decoded['raw'], dict)
|
||||
assert decoded['raw']['raw'] == 'not-a-remote-id-payload'
|
||||
@@ -1,228 +0,0 @@
|
||||
"""Tests for Drone Ops API routes."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
import utils.database as db_mod
|
||||
from utils.drone import get_drone_ops_service
|
||||
|
||||
|
||||
def _set_identity(client, role: str, username: str = 'tester') -> None:
|
||||
with client.session_transaction() as sess:
|
||||
sess['logged_in'] = True
|
||||
sess['role'] = role
|
||||
sess['username'] = username
|
||||
|
||||
|
||||
def _clear_drone_tables() -> None:
|
||||
with db_mod.get_db() as conn:
|
||||
conn.execute('DELETE FROM action_audit_log')
|
||||
conn.execute('DELETE FROM action_approvals')
|
||||
conn.execute('DELETE FROM action_requests')
|
||||
conn.execute('DELETE FROM evidence_manifests')
|
||||
conn.execute('DELETE FROM drone_incident_artifacts')
|
||||
conn.execute('DELETE FROM drone_tracks')
|
||||
conn.execute('DELETE FROM drone_correlations')
|
||||
conn.execute('DELETE FROM drone_detections')
|
||||
conn.execute('DELETE FROM drone_incidents')
|
||||
conn.execute('DELETE FROM drone_sessions')
|
||||
|
||||
|
||||
@pytest.fixture(scope='module', autouse=True)
|
||||
def isolated_drone_db(tmp_path_factory):
|
||||
original_db_dir = db_mod.DB_DIR
|
||||
original_db_path = db_mod.DB_PATH
|
||||
|
||||
tmp_dir = tmp_path_factory.mktemp('drone_ops_db')
|
||||
db_mod.DB_DIR = tmp_dir
|
||||
db_mod.DB_PATH = tmp_dir / 'test_intercept.db'
|
||||
|
||||
if hasattr(db_mod._local, 'connection') and db_mod._local.connection is not None:
|
||||
db_mod._local.connection.close()
|
||||
db_mod._local.connection = None
|
||||
|
||||
db_mod.init_db()
|
||||
yield
|
||||
|
||||
db_mod.close_db()
|
||||
db_mod.DB_DIR = original_db_dir
|
||||
db_mod.DB_PATH = original_db_path
|
||||
db_mod._local.connection = None
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def clean_drone_state():
|
||||
db_mod.init_db()
|
||||
_clear_drone_tables()
|
||||
get_drone_ops_service().disarm_actions(actor='test-reset', reason='test setup')
|
||||
yield
|
||||
_clear_drone_tables()
|
||||
get_drone_ops_service().disarm_actions(actor='test-reset', reason='test teardown')
|
||||
|
||||
|
||||
def test_start_session_requires_operator_role(client):
|
||||
_set_identity(client, role='viewer')
|
||||
response = client.post('/drone-ops/session/start', json={'mode': 'passive'})
|
||||
assert response.status_code == 403
|
||||
data = response.get_json()
|
||||
assert data['required_role'] == 'operator'
|
||||
|
||||
|
||||
def test_session_lifecycle_and_status(client):
|
||||
_set_identity(client, role='operator', username='op1')
|
||||
|
||||
started = client.post('/drone-ops/session/start', json={'mode': 'passive'})
|
||||
assert started.status_code == 200
|
||||
start_data = started.get_json()
|
||||
assert start_data['status'] == 'success'
|
||||
assert start_data['session']['mode'] == 'passive'
|
||||
assert start_data['session']['active'] is True
|
||||
|
||||
status = client.get('/drone-ops/status')
|
||||
assert status.status_code == 200
|
||||
status_data = status.get_json()
|
||||
assert status_data['status'] == 'success'
|
||||
assert status_data['active_session'] is not None
|
||||
assert status_data['active_session']['id'] == start_data['session']['id']
|
||||
|
||||
stopped = client.post('/drone-ops/session/stop', json={'id': start_data['session']['id']})
|
||||
assert stopped.status_code == 200
|
||||
stop_data = stopped.get_json()
|
||||
assert stop_data['status'] == 'success'
|
||||
assert stop_data['session']['active'] is False
|
||||
|
||||
|
||||
def test_detection_ingest_visible_via_endpoint(client):
|
||||
_set_identity(client, role='operator', username='op1')
|
||||
start_resp = client.post('/drone-ops/session/start', json={'mode': 'passive'})
|
||||
assert start_resp.status_code == 200
|
||||
|
||||
service = get_drone_ops_service()
|
||||
service.ingest_event(
|
||||
mode='wifi',
|
||||
event={
|
||||
'bssid': '60:60:1F:AA:BB:CC',
|
||||
'ssid': 'DJI-OPS-TEST',
|
||||
},
|
||||
event_type='network_update',
|
||||
)
|
||||
|
||||
_set_identity(client, role='viewer', username='viewer1')
|
||||
response = client.get('/drone-ops/detections?source=wifi&min_confidence=0.5')
|
||||
assert response.status_code == 200
|
||||
data = response.get_json()
|
||||
assert data['status'] == 'success'
|
||||
assert data['count'] >= 1
|
||||
detection = data['detections'][0]
|
||||
assert detection['source'] == 'wifi'
|
||||
assert detection['confidence'] >= 0.5
|
||||
|
||||
|
||||
def test_incident_artifact_and_manifest_flow(client):
|
||||
_set_identity(client, role='operator', username='op1')
|
||||
created = client.post(
|
||||
'/drone-ops/incidents',
|
||||
json={'title': 'Unidentified UAS', 'severity': 'high'},
|
||||
)
|
||||
assert created.status_code == 201
|
||||
incident = created.get_json()['incident']
|
||||
incident_id = incident['id']
|
||||
|
||||
artifact_resp = client.post(
|
||||
f'/drone-ops/incidents/{incident_id}/artifacts',
|
||||
json={'artifact_type': 'detection', 'artifact_ref': '12345'},
|
||||
)
|
||||
assert artifact_resp.status_code == 201
|
||||
|
||||
_set_identity(client, role='analyst', username='analyst1')
|
||||
manifest_resp = client.post(f'/drone-ops/evidence/{incident_id}/manifest', json={})
|
||||
assert manifest_resp.status_code == 201
|
||||
manifest = manifest_resp.get_json()['manifest']
|
||||
assert manifest['manifest']['integrity']['algorithm'] == 'sha256'
|
||||
assert len(manifest['manifest']['integrity']['digest']) == 64
|
||||
|
||||
_set_identity(client, role='viewer', username='viewer1')
|
||||
listed = client.get(f'/drone-ops/evidence/{incident_id}/manifests')
|
||||
assert listed.status_code == 200
|
||||
listed_data = listed.get_json()
|
||||
assert listed_data['count'] == 1
|
||||
assert listed_data['manifests'][0]['id'] == manifest['id']
|
||||
|
||||
|
||||
def test_action_execution_requires_arming_and_two_approvals(client):
|
||||
_set_identity(client, role='operator', username='op1')
|
||||
incident_resp = client.post('/drone-ops/incidents', json={'title': 'Action Gate Test'})
|
||||
incident_id = incident_resp.get_json()['incident']['id']
|
||||
|
||||
request_resp = client.post(
|
||||
'/drone-ops/actions/request',
|
||||
json={
|
||||
'incident_id': incident_id,
|
||||
'action_type': 'wifi_deauth_test',
|
||||
'payload': {'target': 'aa:bb:cc:dd:ee:ff'},
|
||||
},
|
||||
)
|
||||
assert request_resp.status_code == 201
|
||||
request_id = request_resp.get_json()['request']['id']
|
||||
|
||||
not_armed_resp = client.post(f'/drone-ops/actions/execute/{request_id}', json={})
|
||||
assert not_armed_resp.status_code == 403
|
||||
assert 'not armed' in not_armed_resp.get_json()['message'].lower()
|
||||
|
||||
arm_resp = client.post(
|
||||
'/drone-ops/actions/arm',
|
||||
json={'incident_id': incident_id, 'reason': 'controlled test'},
|
||||
)
|
||||
assert arm_resp.status_code == 200
|
||||
assert arm_resp.get_json()['policy']['armed'] is True
|
||||
|
||||
insufficient_resp = client.post(f'/drone-ops/actions/execute/{request_id}', json={})
|
||||
assert insufficient_resp.status_code == 400
|
||||
assert 'insufficient approvals' in insufficient_resp.get_json()['message'].lower()
|
||||
|
||||
_set_identity(client, role='supervisor', username='supervisor-a')
|
||||
approve_one = client.post(f'/drone-ops/actions/approve/{request_id}', json={'decision': 'approved'})
|
||||
assert approve_one.status_code == 200
|
||||
|
||||
_set_identity(client, role='operator', username='op1')
|
||||
still_insufficient = client.post(f'/drone-ops/actions/execute/{request_id}', json={})
|
||||
assert still_insufficient.status_code == 400
|
||||
|
||||
_set_identity(client, role='supervisor', username='supervisor-b')
|
||||
approve_two = client.post(f'/drone-ops/actions/approve/{request_id}', json={'decision': 'approved'})
|
||||
assert approve_two.status_code == 200
|
||||
assert approve_two.get_json()['request']['status'] == 'approved'
|
||||
|
||||
_set_identity(client, role='operator', username='op1')
|
||||
executed = client.post(f'/drone-ops/actions/execute/{request_id}', json={})
|
||||
assert executed.status_code == 200
|
||||
assert executed.get_json()['request']['status'] == 'executed'
|
||||
|
||||
|
||||
def test_passive_action_executes_after_single_approval(client):
|
||||
_set_identity(client, role='operator', username='op1')
|
||||
incident_resp = client.post('/drone-ops/incidents', json={'title': 'Passive Action Test'})
|
||||
incident_id = incident_resp.get_json()['incident']['id']
|
||||
|
||||
request_resp = client.post(
|
||||
'/drone-ops/actions/request',
|
||||
json={'incident_id': incident_id, 'action_type': 'passive_spectrum_capture'},
|
||||
)
|
||||
request_id = request_resp.get_json()['request']['id']
|
||||
|
||||
arm_resp = client.post(
|
||||
'/drone-ops/actions/arm',
|
||||
json={'incident_id': incident_id, 'reason': 'passive validation'},
|
||||
)
|
||||
assert arm_resp.status_code == 200
|
||||
|
||||
_set_identity(client, role='supervisor', username='supervisor-a')
|
||||
approve_resp = client.post(f'/drone-ops/actions/approve/{request_id}', json={'decision': 'approved'})
|
||||
assert approve_resp.status_code == 200
|
||||
assert approve_resp.get_json()['request']['status'] == 'approved'
|
||||
|
||||
_set_identity(client, role='operator', username='op1')
|
||||
execute_resp = client.post(f'/drone-ops/actions/execute/{request_id}', json={})
|
||||
assert execute_resp.status_code == 200
|
||||
assert execute_resp.get_json()['request']['status'] == 'executed'
|
||||
Reference in New Issue
Block a user