52c9d2b009
Adds PolicyActionHold so an admin can suspend specific actions on a computer without editing the policy itself, covering the break-glass case where a machine must be quiesced immediately. Holds cover multiple actions in one atomic submission rather than one row per action, so a partial failure cannot leave a half-applied hold. Duration is free-form with 0 meaning indefinite (migration 0006 makes expiresat nullable); the prior fixed ceiling was removed because an admin quiescing a misbehaving machine does not know in advance how long it needs, and a hold silently expiring is worse than one that must be cleared deliberately. The What This Does explainer reflects indefinite holds. collect_now exemption: refreshing status was denied on any endpoint with a non-empty effective policy even when the caller held the matching grant, because policy intersection silently removed the action. collect_now is a read-only status refresh with no side effects, so it now sits in POLICY_EXEMPT_ACTIONS alongside the other routine actions. Verified against a synthetic non-empty policy: operators can refresh status again while every write action stays gated.
886 lines
37 KiB
Python
886 lines
37 KiB
Python
"""Admin break-glass over policy gating, and the holds that make it stick.
|
|
|
|
Policy gating is right almost always and wrong at 3am. These tests pin the
|
|
override's boundaries: it must be explicit, admin-only, reasoned, audited, and
|
|
it must NOT become a way to acquire an elevated permission the caller lacks.
|
|
|
|
The hold half exists because an override on a desired-mode action is a fight
|
|
with the reconciler - without suppression the next check-in re-enqueues the
|
|
opposite command and the override looks broken.
|
|
"""
|
|
|
|
from datetime import datetime, timedelta
|
|
|
|
from werkzeug.security import generate_password_hash
|
|
|
|
from shopdb.core.models import Setting
|
|
|
|
|
|
def _enable_policies(db):
|
|
if not Setting.query.filter_by(key='computers_configpolicies_enabled').first():
|
|
db.session.add(Setting(key='computers_configpolicies_enabled',
|
|
value='true'))
|
|
db.session.commit()
|
|
|
|
|
|
def _endpoint(db, hostname='holdbox'):
|
|
"""An Asset + Computer pair, the minimum a policy can resolve against."""
|
|
from shopdb.api import Asset, AssetType
|
|
from plugins.computers.models import Computer
|
|
|
|
assettype = AssetType.query.filter_by(assettype='computer').first()
|
|
if assettype is None:
|
|
assettype = AssetType(assettype='computer', pluginname='computers',
|
|
tablename='computers')
|
|
db.session.add(assettype)
|
|
db.session.flush()
|
|
|
|
asset = Asset(assetnumber=hostname, name=hostname,
|
|
assettypeid=assettype.assettypeid, isactive=True)
|
|
db.session.add(asset)
|
|
db.session.flush()
|
|
db.session.add(Computer(assetid=asset.assetid, hostname=hostname))
|
|
# The enqueue route 404s without a reported state, so the endpoint has to
|
|
# look like it has checked in at least once.
|
|
from plugins.endpoints.models import EndpointState
|
|
db.session.add(EndpointState(assetid=asset.assetid, hostname=hostname))
|
|
db.session.commit()
|
|
return asset.assetid
|
|
|
|
|
|
def _login(client, username, password='testpass'):
|
|
response = client.post('/api/auth/login',
|
|
json={'username': username, 'password': password})
|
|
token = response.get_json()['data']['access_token']
|
|
return {'Authorization': f'Bearer {token}'}
|
|
|
|
|
|
def _user(db, client, username, permnames, admin=False):
|
|
from shopdb.core.models import User, Role, Permission
|
|
|
|
Permission.seed()
|
|
Permission.seed_entries([
|
|
('endpoints.restart', 'Remote restart', 'endpoints'),
|
|
])
|
|
db.session.flush()
|
|
role = Role(rolename='admin' if admin else username + 'role')
|
|
existing = Role.query.filter_by(rolename=role.rolename).first()
|
|
if existing is not None:
|
|
role = existing
|
|
else:
|
|
db.session.add(role)
|
|
db.session.flush()
|
|
role.permissions = Permission.query.filter(
|
|
Permission.name.in_(permnames)).all()
|
|
user = User(username=username, email=username + '@test.local',
|
|
passwordhash=generate_password_hash('testpass'))
|
|
user.roles.append(role)
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
return _login(client, username)
|
|
|
|
|
|
def _deny_everything(db, assetid):
|
|
"""A policy that matches this endpoint but permits only an unrelated action,
|
|
so the action under test is denied by gating."""
|
|
from plugins.computers.models import (ConfigurationPolicy,
|
|
ConfigurationPolicyAction,
|
|
BusinessPolicy, BusinessPolicyItem)
|
|
policy = ConfigurationPolicy(name='Locked down', scopejson=None)
|
|
db.session.add(policy)
|
|
db.session.flush()
|
|
db.session.add(ConfigurationPolicyAction(
|
|
configurationpolicyid=policy.configurationpolicyid,
|
|
action='mtconnect_status'))
|
|
business = BusinessPolicy(name='Fleet')
|
|
db.session.add(business)
|
|
db.session.flush()
|
|
db.session.add(BusinessPolicyItem(
|
|
businesspolicyid=business.businesspolicyid,
|
|
configurationpolicyid=policy.configurationpolicyid))
|
|
db.session.commit()
|
|
|
|
|
|
def _url(assetid):
|
|
return f'/api/endpoints/{assetid}/commands'
|
|
|
|
|
|
def test_policy_denial_still_applies_without_an_override(client, db):
|
|
"""The default path must be unchanged: no flag, no bypass."""
|
|
_enable_policies(db)
|
|
assetid = _endpoint(db)
|
|
_deny_everything(db, assetid)
|
|
headers = _user(db, client, 'plainadmin',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
|
|
response = client.post(_url(assetid), headers=headers,
|
|
json={'action': 'service_status',
|
|
'target': 'spooler'})
|
|
assert response.status_code == 400, response.get_json()
|
|
assert 'not permitted by policy' in \
|
|
response.get_json()['data']['error']['message']
|
|
|
|
|
|
def test_a_non_admin_cannot_override_a_policy_denial(client, db):
|
|
"""endpoints.command is enough to queue work, never enough to break glass."""
|
|
_enable_policies(db)
|
|
assetid = _endpoint(db)
|
|
_deny_everything(db, assetid)
|
|
headers = _user(db, client, 'operator',
|
|
['endpoints.view', 'endpoints.command'])
|
|
|
|
response = client.post(_url(assetid), headers=headers,
|
|
json={'action': 'service_status',
|
|
'target': 'spooler',
|
|
'policyoverride': True,
|
|
'overridereason': 'trying it on'})
|
|
assert response.status_code == 403, response.get_json()
|
|
assert 'admin' in response.get_json()['data']['error']['message']
|
|
|
|
|
|
def test_an_override_without_a_reason_is_refused(client, db):
|
|
"""An unexplained break-glass is an unreviewable one."""
|
|
_enable_policies(db)
|
|
assetid = _endpoint(db)
|
|
_deny_everything(db, assetid)
|
|
headers = _user(db, client, 'reasonless',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
|
|
response = client.post(_url(assetid), headers=headers,
|
|
json={'action': 'service_status',
|
|
'target': 'spooler',
|
|
'policyoverride': True})
|
|
assert response.status_code == 400, response.get_json()
|
|
assert 'overridereason' in response.get_json()['data']['error']['message']
|
|
|
|
|
|
def test_an_admin_override_is_accepted_and_audited(client, db):
|
|
"""The whole point, plus the audit trail that justifies allowing it."""
|
|
from shopdb.api import AuditLog
|
|
|
|
_enable_policies(db)
|
|
assetid = _endpoint(db)
|
|
_deny_everything(db, assetid)
|
|
headers = _user(db, client, 'breakglass',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
|
|
response = client.post(_url(assetid), headers=headers,
|
|
json={'action': 'service_status',
|
|
'target': 'spooler',
|
|
'policyoverride': True,
|
|
'overridereason': 'line down, need status'})
|
|
assert response.status_code in (200, 201), response.get_json()
|
|
|
|
entry = AuditLog.query.filter_by(action='policyoverridden').first()
|
|
assert entry is not None, 'break-glass must leave an audit record'
|
|
|
|
|
|
def test_override_is_checked_after_the_elevated_permission_gate(client, db):
|
|
"""The override widens WHERE an action may run, never WHICH actions the
|
|
caller may use.
|
|
|
|
A human admin holds every permission via the admin bypass, so this is moot
|
|
for people and cannot be shown with an admin fixture. It matters for TOKENS:
|
|
authz suspends the admin bypass for a scoped personal token, so a token
|
|
scoped to endpoints.command is refused at the elevated gate and never
|
|
reaches the override branch at all - even though its owner is an admin, and
|
|
even though it asks for the override explicitly.
|
|
"""
|
|
from shopdb.core.models import Permission
|
|
|
|
_enable_policies(db)
|
|
assetid = _endpoint(db, 'tokenbox')
|
|
_deny_everything(db, assetid)
|
|
Permission.seed_entries([
|
|
('apitokens.create', 'Mint tokens', 'core'),
|
|
])
|
|
db.session.commit()
|
|
ownerheaders = _user(db, client, 'tokenadmin',
|
|
['endpoints.view', 'endpoints.command',
|
|
'apitokens.create'], admin=True)
|
|
|
|
minted = client.post('/api/apitokens', headers=ownerheaders,
|
|
json={'name': 'agent',
|
|
'scopes': ['endpoints.command']})
|
|
secret = minted.get_json()['data']['secret']
|
|
headers = {'Authorization': f'Bearer {secret}'}
|
|
|
|
response = client.post(_url(assetid), headers=headers,
|
|
json={'action': 'system_restart',
|
|
'policyoverride': True,
|
|
'overridereason': 'leaked token tries it on'})
|
|
assert response.status_code == 403, response.get_json()
|
|
|
|
|
|
def test_a_hold_suppresses_desired_state_reconciliation(client, db):
|
|
"""Without this the reconciler undoes the override on the next check-in."""
|
|
from plugins.computers.models import PolicyActionHold
|
|
from plugins.computers.services.policyresolver import effective_policy_for
|
|
|
|
_enable_policies(db)
|
|
assetid = _endpoint(db, 'desiredbox')
|
|
|
|
from plugins.computers.models import (ConfigurationPolicy,
|
|
ConfigurationPolicyAction,
|
|
BusinessPolicy, BusinessPolicyItem)
|
|
policy = ConfigurationPolicy(name='Keep it running', scopejson=None)
|
|
db.session.add(policy)
|
|
db.session.flush()
|
|
db.session.add(ConfigurationPolicyAction(
|
|
configurationpolicyid=policy.configurationpolicyid,
|
|
action='service_start', mode='desired',
|
|
desiredstate={'running': True}))
|
|
business = BusinessPolicy(name='Desired fleet')
|
|
db.session.add(business)
|
|
db.session.flush()
|
|
db.session.add(BusinessPolicyItem(
|
|
businesspolicyid=business.businesspolicyid,
|
|
configurationpolicyid=policy.configurationpolicyid))
|
|
db.session.commit()
|
|
|
|
assert 'service_start' in effective_policy_for(assetid).desired_actions()
|
|
|
|
db.session.add(PolicyActionHold(
|
|
assetid=assetid, action='service_start',
|
|
expiresat=datetime.utcnow() + timedelta(minutes=60),
|
|
reason='maintenance window'))
|
|
db.session.commit()
|
|
|
|
effective = effective_policy_for(assetid)
|
|
assert 'service_start' not in effective.desired_actions()
|
|
# A hold suspends drift correction; it does not revoke the permission to
|
|
# run the action by hand.
|
|
assert 'service_start' in effective.allowed_actions()
|
|
|
|
|
|
def test_an_expired_hold_stops_suppressing(client, db):
|
|
"""Policy resumes on its own. Nobody has to remember to clear a hold, and a
|
|
stalled sweeper cannot silently disable reconciliation forever."""
|
|
from plugins.computers.models import PolicyActionHold
|
|
from plugins.computers.services.policyresolver import effective_policy_for
|
|
|
|
_enable_policies(db)
|
|
assetid = _endpoint(db, 'expiredbox')
|
|
|
|
from plugins.computers.models import (ConfigurationPolicy,
|
|
ConfigurationPolicyAction,
|
|
BusinessPolicy, BusinessPolicyItem)
|
|
policy = ConfigurationPolicy(name='Still running', scopejson=None)
|
|
db.session.add(policy)
|
|
db.session.flush()
|
|
db.session.add(ConfigurationPolicyAction(
|
|
configurationpolicyid=policy.configurationpolicyid,
|
|
action='service_start', mode='desired',
|
|
desiredstate={'running': True}))
|
|
business = BusinessPolicy(name='Expiry fleet')
|
|
db.session.add(business)
|
|
db.session.flush()
|
|
db.session.add(BusinessPolicyItem(
|
|
businesspolicyid=business.businesspolicyid,
|
|
configurationpolicyid=policy.configurationpolicyid))
|
|
db.session.add(PolicyActionHold(
|
|
assetid=assetid, action='service_start',
|
|
expiresat=datetime.utcnow() - timedelta(minutes=1),
|
|
reason='window that already closed'))
|
|
db.session.commit()
|
|
|
|
assert 'service_start' in effective_policy_for(assetid).desired_actions()
|
|
|
|
|
|
def test_active_holds_are_visible_to_a_non_admin_viewer(client, db):
|
|
"""A hold explains drift, so it must not be admin-only reading.
|
|
|
|
An operator who can see the machine but not the hold is left debugging
|
|
suppressed reconciliation as if it were a fault.
|
|
"""
|
|
from plugins.computers.models import PolicyActionHold
|
|
|
|
assetid = _endpoint(db)
|
|
db.session.add(PolicyActionHold(
|
|
assetid=assetid, action='service_stop',
|
|
expiresat=datetime.utcnow() + timedelta(minutes=60),
|
|
createdbyusername='nightadmin',
|
|
reason='vendor on site'))
|
|
db.session.commit()
|
|
|
|
headers = _user(db, client, 'viewer', ['endpoints.view'])
|
|
response = client.get(f'/api/endpoints/{assetid}/policyholds',
|
|
headers=headers)
|
|
assert response.status_code == 200, response.get_json()
|
|
holds = response.get_json()['data']
|
|
assert len(holds) == 1
|
|
assert holds[0]['action'] == 'service_stop'
|
|
assert holds[0]['createdbyusername'] == 'nightadmin'
|
|
assert holds[0]['reason'] == 'vendor on site'
|
|
|
|
|
|
def test_expired_holds_are_not_reported_as_active(client, db):
|
|
"""Expiry is evaluated on read, matching the resolver.
|
|
|
|
If the route reported expired rows the banner would claim enforcement is
|
|
paused long after it resumed - worse than showing nothing.
|
|
"""
|
|
from plugins.computers.models import PolicyActionHold
|
|
|
|
assetid = _endpoint(db)
|
|
db.session.add(PolicyActionHold(
|
|
assetid=assetid, action='service_stop',
|
|
expiresat=datetime.utcnow() - timedelta(minutes=1),
|
|
createdbyusername='nightadmin',
|
|
reason='window that already closed'))
|
|
db.session.commit()
|
|
|
|
headers = _user(db, client, 'viewer2', ['endpoints.view'])
|
|
response = client.get(f'/api/endpoints/{assetid}/policyholds',
|
|
headers=headers)
|
|
assert response.status_code == 200, response.get_json()
|
|
assert response.get_json()['data'] == []
|
|
|
|
|
|
def _desired_policy(db, assetid, action='service_start'):
|
|
"""A policy putting one action under desired-state enforcement."""
|
|
from plugins.computers.models import (ConfigurationPolicy,
|
|
ConfigurationPolicyAction,
|
|
BusinessPolicy, BusinessPolicyItem)
|
|
policy = ConfigurationPolicy(name=f'Desired {action}', scopejson=None)
|
|
db.session.add(policy)
|
|
db.session.flush()
|
|
db.session.add(ConfigurationPolicyAction(
|
|
configurationpolicyid=policy.configurationpolicyid,
|
|
action=action, mode='desired', desiredstate={'running': True}))
|
|
business = BusinessPolicy(name=f'Fleet {action}')
|
|
db.session.add(business)
|
|
db.session.flush()
|
|
db.session.add(BusinessPolicyItem(
|
|
businesspolicyid=business.businesspolicyid,
|
|
configurationpolicyid=policy.configurationpolicyid))
|
|
db.session.commit()
|
|
|
|
|
|
def test_an_admin_can_place_a_hold_directly(client, db):
|
|
"""The hold must be reachable deliberately, not only by tripping a denial."""
|
|
_enable_policies(db)
|
|
assetid = _endpoint(db)
|
|
_desired_policy(db, assetid)
|
|
headers = _user(db, client, 'holdadmin',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
|
|
response = client.post(f'/api/endpoints/{assetid}/policyholds',
|
|
headers=headers,
|
|
json={'action': 'service_start',
|
|
'reason': 'vendor on site',
|
|
'minutes': 60})
|
|
assert response.status_code == 200, response.get_json()
|
|
holds = response.get_json()['data']
|
|
assert [h['action'] for h in holds] == ['service_start']
|
|
|
|
# The point of a hold: reconciliation stops asking for this action.
|
|
from plugins.computers.services.policyresolver import effective_policy_for
|
|
assert 'service_start' not in effective_policy_for(assetid).desired_actions()
|
|
|
|
|
|
def test_a_non_admin_cannot_place_a_hold(client, db):
|
|
"""endpoints.command queues work; it does not switch off enforcement."""
|
|
_enable_policies(db)
|
|
assetid = _endpoint(db)
|
|
_desired_policy(db, assetid)
|
|
headers = _user(db, client, 'holdoperator',
|
|
['endpoints.view', 'endpoints.command'])
|
|
|
|
response = client.post(f'/api/endpoints/{assetid}/policyholds',
|
|
headers=headers,
|
|
json={'action': 'service_start',
|
|
'reason': 'trying it on'})
|
|
assert response.status_code == 403, response.get_json()
|
|
|
|
|
|
def test_a_hold_requires_a_reason(client, db):
|
|
"""An unexplained hold is an unattributable outage waiting to happen."""
|
|
_enable_policies(db)
|
|
assetid = _endpoint(db)
|
|
_desired_policy(db, assetid)
|
|
headers = _user(db, client, 'holdadmin2',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
|
|
response = client.post(f'/api/endpoints/{assetid}/policyholds',
|
|
headers=headers,
|
|
json={'action': 'service_start', 'reason': ' '})
|
|
assert response.status_code == 400, response.get_json()
|
|
|
|
|
|
def test_holding_a_non_desired_action_is_refused(client, db):
|
|
"""A hold on a gate-mode action would suppress nothing.
|
|
|
|
Accepting it would report a hold that has no effect, which is worse than
|
|
refusing outright.
|
|
"""
|
|
_enable_policies(db)
|
|
assetid = _endpoint(db)
|
|
_deny_everything(db, assetid)
|
|
headers = _user(db, client, 'holdadmin3',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
|
|
response = client.post(f'/api/endpoints/{assetid}/policyholds',
|
|
headers=headers,
|
|
json={'action': 'mtconnect_status',
|
|
'reason': 'not under enforcement'})
|
|
assert response.status_code == 400, response.get_json()
|
|
assert 'nothing to hold' in \
|
|
response.get_json()['data']['error']['message']
|
|
|
|
|
|
def test_an_admin_can_release_a_hold_early(client, db):
|
|
"""Release resumes reconciliation without waiting for expiry."""
|
|
_enable_policies(db)
|
|
assetid = _endpoint(db)
|
|
_desired_policy(db, assetid)
|
|
headers = _user(db, client, 'holdadmin4',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
|
|
client.post(f'/api/endpoints/{assetid}/policyholds', headers=headers,
|
|
json={'action': 'service_start', 'reason': 'swapping a disk'})
|
|
|
|
from plugins.computers.services.policyresolver import effective_policy_for
|
|
assert 'service_start' not in effective_policy_for(assetid).desired_actions()
|
|
|
|
response = client.delete(
|
|
f'/api/endpoints/{assetid}/policyholds/service_start', headers=headers)
|
|
assert response.status_code == 200, response.get_json()
|
|
assert response.get_json()['data'] == []
|
|
assert 'service_start' in effective_policy_for(assetid).desired_actions()
|
|
|
|
|
|
def test_releasing_a_hold_keeps_the_audit_row(client, db):
|
|
"""Release soft-deletes: the hold IS the record of what was suppressed."""
|
|
from plugins.computers.models import PolicyActionHold
|
|
|
|
_enable_policies(db)
|
|
assetid = _endpoint(db)
|
|
_desired_policy(db, assetid)
|
|
headers = _user(db, client, 'holdadmin5',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
|
|
client.post(f'/api/endpoints/{assetid}/policyholds', headers=headers,
|
|
json={'action': 'service_start', 'reason': 'keep this record'})
|
|
client.delete(f'/api/endpoints/{assetid}/policyholds/service_start',
|
|
headers=headers)
|
|
|
|
row = PolicyActionHold.query.filter_by(assetid=assetid,
|
|
action='service_start').first()
|
|
assert row is not None
|
|
assert row.isactive is False
|
|
assert row.reason == 'keep this record'
|
|
|
|
|
|
def test_holdable_actions_lists_desired_actions_with_hold_state(client, db):
|
|
"""The picker must keep showing an action after it is held.
|
|
|
|
Built from desired_actions() it would vanish on being held, making the
|
|
current state unreadable.
|
|
"""
|
|
_enable_policies(db)
|
|
assetid = _endpoint(db)
|
|
_desired_policy(db, assetid)
|
|
headers = _user(db, client, 'holdadmin6',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
|
|
response = client.get(f'/api/endpoints/{assetid}/holdableactions',
|
|
headers=headers)
|
|
assert response.status_code == 200, response.get_json()
|
|
assert response.get_json()['data'] == [
|
|
{'action': 'service_start', 'held': False}]
|
|
|
|
client.post(f'/api/endpoints/{assetid}/policyholds', headers=headers,
|
|
json={'action': 'service_start', 'reason': 'still listed'})
|
|
|
|
response = client.get(f'/api/endpoints/{assetid}/holdableactions',
|
|
headers=headers)
|
|
assert response.get_json()['data'] == [
|
|
{'action': 'service_start', 'held': True}]
|
|
|
|
|
|
def _lock_url(assetid):
|
|
return f'/api/endpoints/{assetid}/lock'
|
|
|
|
|
|
def test_a_lock_refuses_a_non_admin_command(client, db):
|
|
"""The whole point: a locked endpoint stops other people acting on it.
|
|
|
|
No policy is enabled here, so nothing but the lock can be doing the
|
|
refusing - otherwise this test would pass for the wrong reason.
|
|
"""
|
|
assetid = _endpoint(db)
|
|
adminheaders = _user(db, client, 'lockadmin',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
client.post(_lock_url(assetid), headers=adminheaders,
|
|
json={'reason': 'mid repair'})
|
|
|
|
headers = _user(db, client, 'lockoperator',
|
|
['endpoints.view', 'endpoints.command'])
|
|
response = client.post(_url(assetid), headers=headers,
|
|
json={'action': 'service_status',
|
|
'target': 'spooler'})
|
|
assert response.status_code == 403, response.get_json()
|
|
|
|
|
|
def test_a_lock_refusal_names_the_owner_and_reason(client, db):
|
|
"""A bare "forbidden" sends an operator hunting through policy for a cause
|
|
that is not there. The refusal has to say who to ask and why."""
|
|
assetid = _endpoint(db)
|
|
adminheaders = _user(db, client, 'lockadmin2',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
client.post(_lock_url(assetid), headers=adminheaders,
|
|
json={'reason': 'validated part running'})
|
|
|
|
headers = _user(db, client, 'lockoperator2',
|
|
['endpoints.view', 'endpoints.command'])
|
|
response = client.post(_url(assetid), headers=headers,
|
|
json={'action': 'service_status',
|
|
'target': 'spooler'})
|
|
message = response.get_json()['data']['error']['message']
|
|
assert 'lockadmin2' in message
|
|
assert 'validated part running' in message
|
|
|
|
|
|
def test_an_admin_acts_through_a_lock_without_an_override(client, db):
|
|
"""The lock gates WHO may act, not WHICH actions exist, so there is no
|
|
policy decision for an admin to break past. Requiring break-glass here would
|
|
mean an admin overriding their own quarantine to do the repair it was set
|
|
for."""
|
|
assetid = _endpoint(db)
|
|
headers = _user(db, client, 'lockadmin3',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
client.post(_lock_url(assetid), headers=headers,
|
|
json={'reason': 'mine to work on'})
|
|
|
|
response = client.post(_url(assetid), headers=headers,
|
|
json={'action': 'service_status',
|
|
'target': 'spooler'})
|
|
assert response.status_code == 201, response.get_json()
|
|
|
|
|
|
def test_a_lock_cannot_be_broken_by_a_policy_override(client, db):
|
|
"""The two gates are independent and the lock is the outer one.
|
|
|
|
policyoverride exists to push past a POLICY denial. If it also dissolved a
|
|
lock, any non-admin who learned the flag could act on a quarantined machine,
|
|
and the lock would be advisory.
|
|
"""
|
|
_enable_policies(db)
|
|
assetid = _endpoint(db)
|
|
_deny_everything(db, assetid)
|
|
adminheaders = _user(db, client, 'lockadmin4',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
client.post(_lock_url(assetid), headers=adminheaders,
|
|
json={'reason': 'do not touch'})
|
|
|
|
headers = _user(db, client, 'lockoperator4',
|
|
['endpoints.view', 'endpoints.command'])
|
|
response = client.post(_url(assetid), headers=headers,
|
|
json={'action': 'service_status',
|
|
'target': 'spooler',
|
|
'policyoverride': True,
|
|
'overridereason': 'trying to slip past'})
|
|
assert response.status_code == 403, response.get_json()
|
|
assert 'locked' in response.get_json()['data']['error']['message'].lower()
|
|
|
|
|
|
def test_a_non_admin_cannot_place_a_lock(client, db):
|
|
assetid = _endpoint(db)
|
|
headers = _user(db, client, 'lockoperator5',
|
|
['endpoints.view', 'endpoints.command'])
|
|
|
|
response = client.post(_lock_url(assetid), headers=headers,
|
|
json={'reason': 'not my call'})
|
|
assert response.status_code == 403, response.get_json()
|
|
|
|
|
|
def test_a_lock_requires_a_reason(client, db):
|
|
"""The reason is the message a blocked operator reads, not paperwork."""
|
|
assetid = _endpoint(db)
|
|
headers = _user(db, client, 'lockadmin6',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
|
|
response = client.post(_lock_url(assetid), headers=headers,
|
|
json={'reason': ' '})
|
|
assert response.status_code == 400, response.get_json()
|
|
|
|
|
|
def test_locking_twice_does_not_stack_a_second_lock(client, db):
|
|
"""Two active locks on one endpoint is not a state the table should hold,
|
|
and the caller who locked a locked machine got what they asked for."""
|
|
assetid = _endpoint(db)
|
|
headers = _user(db, client, 'lockadmin7',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
|
|
first = client.post(_lock_url(assetid), headers=headers,
|
|
json={'reason': 'first'})
|
|
second = client.post(_lock_url(assetid), headers=headers,
|
|
json={'reason': 'second'})
|
|
assert second.status_code == 200, second.get_json()
|
|
# The original stands - the second call did not silently replace the reason
|
|
# the first admin recorded.
|
|
assert second.get_json()['data']['endpointlockid'] == \
|
|
first.get_json()['data']['endpointlockid']
|
|
assert second.get_json()['data']['reason'] == 'first'
|
|
|
|
|
|
def test_releasing_a_lock_restores_non_admin_commands(client, db):
|
|
assetid = _endpoint(db)
|
|
adminheaders = _user(db, client, 'lockadmin8',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
client.post(_lock_url(assetid), headers=adminheaders,
|
|
json={'reason': 'briefly'})
|
|
client.delete(_lock_url(assetid), headers=adminheaders)
|
|
|
|
headers = _user(db, client, 'lockoperator8',
|
|
['endpoints.view', 'endpoints.command'])
|
|
response = client.post(_url(assetid), headers=headers,
|
|
json={'action': 'service_status',
|
|
'target': 'spooler'})
|
|
assert response.status_code == 201, response.get_json()
|
|
|
|
|
|
def test_releasing_a_lock_keeps_the_row_as_history(client, db):
|
|
"""Released by stamping releasedat, not by deleting. The record of who
|
|
quarantined a machine is most wanted after the lock is gone and no longer
|
|
there to explain itself."""
|
|
from plugins.endpoints.models import EndpointLock
|
|
|
|
assetid = _endpoint(db)
|
|
headers = _user(db, client, 'lockadmin9',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
client.post(_lock_url(assetid), headers=headers,
|
|
json={'reason': 'keep this record'})
|
|
client.delete(_lock_url(assetid), headers=headers)
|
|
|
|
row = EndpointLock.query.filter_by(assetid=assetid).first()
|
|
assert row is not None
|
|
assert row.reason == 'keep this record'
|
|
assert row.lockedbyusername == 'lockadmin9'
|
|
assert row.releasedat is not None
|
|
assert row.releasedbyusername == 'lockadmin9'
|
|
|
|
|
|
def test_releasing_an_unlocked_endpoint_is_a_404(client, db):
|
|
"""So a caller can tell "unlocked it" from "there was nothing to unlock"."""
|
|
assetid = _endpoint(db)
|
|
headers = _user(db, client, 'lockadmin10',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
|
|
response = client.delete(_lock_url(assetid), headers=headers)
|
|
assert response.status_code == 404, response.get_json()
|
|
|
|
|
|
def test_a_non_admin_can_read_the_lock(client, db):
|
|
"""Read-gated, not admin-gated. An operator who cannot see the lock is left
|
|
with a bare refusal and files a ticket about a machine that is working."""
|
|
assetid = _endpoint(db)
|
|
adminheaders = _user(db, client, 'lockadmin11',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
client.post(_lock_url(assetid), headers=adminheaders,
|
|
json={'reason': 'visible to all'})
|
|
|
|
headers = _user(db, client, 'lockoperator11', ['endpoints.view'])
|
|
response = client.get(_lock_url(assetid), headers=headers)
|
|
assert response.status_code == 200, response.get_json()
|
|
assert response.get_json()['data']['reason'] == 'visible to all'
|
|
|
|
|
|
def test_an_unlocked_endpoint_reports_no_lock(client, db):
|
|
assetid = _endpoint(db)
|
|
headers = _user(db, client, 'lockoperator12', ['endpoints.view'])
|
|
|
|
response = client.get(_lock_url(assetid), headers=headers)
|
|
assert response.status_code == 200, response.get_json()
|
|
# success_response(None) omits data entirely rather than emitting null, so
|
|
# "no lock" is an absent key. The frontend reads it as response.data?.data
|
|
# || null, which lands on the same answer either way.
|
|
assert response.get_json().get('data') is None
|
|
|
|
|
|
def test_a_lock_does_not_suppress_policy_reconciliation(client, db):
|
|
"""The two axes stay separate. A lock gates callers; it must not reach into
|
|
the resolver and start suppressing desired-state actions, which is the hold's
|
|
job and carries an expiry a lock does not have."""
|
|
from plugins.computers.services.policyresolver import effective_policy_for
|
|
|
|
_enable_policies(db)
|
|
assetid = _endpoint(db)
|
|
_desired_policy(db, assetid)
|
|
headers = _user(db, client, 'lockadmin13',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
client.post(_lock_url(assetid), headers=headers,
|
|
json={'reason': 'locked but still reconciled'})
|
|
|
|
assert 'service_start' in effective_policy_for(assetid).desired_actions()
|
|
|
|
|
|
def test_an_admin_can_hold_several_actions_at_once(client, db):
|
|
"""One dialog submission covers the whole job, not one action per trip.
|
|
|
|
An admin about to work on a machine usually needs several actions pinned,
|
|
and making them repeat the dialog invites stopping halfway.
|
|
"""
|
|
from plugins.computers.services.policyresolver import effective_policy_for
|
|
|
|
_enable_policies(db)
|
|
assetid = _endpoint(db)
|
|
_desired_policy(db, assetid, action='service_start')
|
|
_desired_policy(db, assetid, action='service_stop')
|
|
headers = _user(db, client, 'multiadmin1',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
|
|
response = client.post(f'/api/endpoints/{assetid}/policyholds',
|
|
headers=headers,
|
|
json={'actions': ['service_start', 'service_stop'],
|
|
'reason': 'stripping the enclosure',
|
|
'minutes': 60})
|
|
assert response.status_code == 200, response.get_json()
|
|
held = {h['action'] for h in response.get_json()['data']}
|
|
assert held == {'service_start', 'service_stop'}
|
|
|
|
desired = effective_policy_for(assetid).desired_actions()
|
|
assert 'service_start' not in desired
|
|
assert 'service_stop' not in desired
|
|
|
|
|
|
def test_a_multi_action_hold_is_all_or_nothing(client, db):
|
|
"""One bad name rejects the request whole.
|
|
|
|
Partial success is the wrong failure mode on a break-glass control: the
|
|
admin walks away believing the machine is pinned, and the reconciler undoes
|
|
the work covered by the holds that never landed.
|
|
"""
|
|
from plugins.computers.models import PolicyActionHold
|
|
|
|
_enable_policies(db)
|
|
assetid = _endpoint(db)
|
|
_desired_policy(db, assetid, action='service_start')
|
|
headers = _user(db, client, 'multiadmin2',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
|
|
response = client.post(
|
|
f'/api/endpoints/{assetid}/policyholds',
|
|
headers=headers,
|
|
# service_start is holdable; mtconnect_status is not under enforcement.
|
|
json={'actions': ['service_start', 'mtconnect_status'],
|
|
'reason': 'half of this is invalid'})
|
|
assert response.status_code == 400, response.get_json()
|
|
|
|
# The valid half must NOT have been written on the way to the refusal.
|
|
assert PolicyActionHold.query.filter_by(assetid=assetid).count() == 0
|
|
|
|
|
|
def test_a_zero_minute_hold_never_expires(client, db):
|
|
"""0 means indefinite: it outlives any window a dated hold would have."""
|
|
from plugins.computers.models import PolicyActionHold
|
|
from plugins.computers.services.policyresolver import effective_policy_for
|
|
|
|
_enable_policies(db)
|
|
assetid = _endpoint(db)
|
|
_desired_policy(db, assetid)
|
|
headers = _user(db, client, 'indefadmin1',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
|
|
response = client.post(f'/api/endpoints/{assetid}/policyholds',
|
|
headers=headers,
|
|
json={'actions': ['service_start'],
|
|
'reason': 'awaiting a replacement part',
|
|
'minutes': 0})
|
|
assert response.status_code == 200, response.get_json()
|
|
|
|
row = PolicyActionHold.query.filter_by(assetid=assetid,
|
|
action='service_start').first()
|
|
assert row.expiresat is None
|
|
# Reported active and honoured by the resolver despite having no expiry.
|
|
assert response.get_json()['data'][0]['expiresat'] is None
|
|
assert 'service_start' not in effective_policy_for(assetid).desired_actions()
|
|
|
|
|
|
def test_an_indefinite_hold_can_still_be_released(client, db):
|
|
"""The one hold that never self-clears must be clearable by hand.
|
|
|
|
Without a null branch in the release query this would be the only kind of
|
|
hold the release button could not touch - permanently stuck.
|
|
"""
|
|
from plugins.computers.services.policyresolver import effective_policy_for
|
|
|
|
_enable_policies(db)
|
|
assetid = _endpoint(db)
|
|
_desired_policy(db, assetid)
|
|
headers = _user(db, client, 'indefadmin2',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
client.post(f'/api/endpoints/{assetid}/policyholds', headers=headers,
|
|
json={'actions': ['service_start'], 'reason': 'indefinite',
|
|
'minutes': 0})
|
|
|
|
response = client.delete(
|
|
f'/api/endpoints/{assetid}/policyholds/service_start', headers=headers)
|
|
assert response.status_code == 200, response.get_json()
|
|
assert 'service_start' in effective_policy_for(assetid).desired_actions()
|
|
|
|
|
|
def test_a_negative_hold_window_is_refused(client, db):
|
|
"""A negative window places an already-expired hold.
|
|
|
|
That reads as success in the UI while suppressing nothing - the silent
|
|
failure the holdable-action check exists to prevent.
|
|
"""
|
|
_enable_policies(db)
|
|
assetid = _endpoint(db)
|
|
_desired_policy(db, assetid)
|
|
headers = _user(db, client, 'indefadmin3',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
|
|
response = client.post(f'/api/endpoints/{assetid}/policyholds',
|
|
headers=headers,
|
|
json={'actions': ['service_start'],
|
|
'reason': 'backwards', 'minutes': -30})
|
|
assert response.status_code == 400, response.get_json()
|
|
|
|
|
|
def test_re_holding_an_active_action_does_not_stack_rows(client, db):
|
|
"""Idempotent, matching the endpoint lock.
|
|
|
|
A duplicate row would have to be released once per hidden copy before
|
|
reconciliation actually resumed.
|
|
"""
|
|
from plugins.computers.models import PolicyActionHold
|
|
from plugins.computers.services.policyresolver import effective_policy_for
|
|
|
|
_enable_policies(db)
|
|
assetid = _endpoint(db)
|
|
_desired_policy(db, assetid)
|
|
headers = _user(db, client, 'dupeadmin1',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
|
|
for _ in range(2):
|
|
client.post(f'/api/endpoints/{assetid}/policyholds', headers=headers,
|
|
json={'actions': ['service_start'], 'reason': 'twice'})
|
|
|
|
assert PolicyActionHold.query.filter_by(assetid=assetid,
|
|
action='service_start').count() == 1
|
|
|
|
# One release is therefore enough to resume reconciliation.
|
|
client.delete(f'/api/endpoints/{assetid}/policyholds/service_start',
|
|
headers=headers)
|
|
assert 'service_start' in effective_policy_for(assetid).desired_actions()
|
|
|
|
|
|
def test_a_single_action_hold_payload_still_works(client, db):
|
|
"""The pre-list payload shape stays accepted so older callers keep working."""
|
|
_enable_policies(db)
|
|
assetid = _endpoint(db)
|
|
_desired_policy(db, assetid)
|
|
headers = _user(db, client, 'legacyadmin1',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
|
|
response = client.post(f'/api/endpoints/{assetid}/policyholds',
|
|
headers=headers,
|
|
json={'action': 'service_start',
|
|
'reason': 'legacy payload'})
|
|
assert response.status_code == 200, response.get_json()
|
|
assert [h['action'] for h in response.get_json()['data']] == \
|
|
['service_start']
|