7480e4fe3a
Per-PC credentials (replaces the 09-14 client-certificate attempt): - POST /api/endpoints/enroll issues a credential to a PC that matches exactly one computer by hostname and BIOS serial and has none yet. Anything else (unknown or duplicate hostname, serial mismatch, already enrolled) waits as a request an admin approves or rejects. The agent's persisted nonce binds an approved request to the PC that asked. - One auth helper serves every agent route. With X-Endpoint-Credential the asset comes from the credential and a request naming another PC gets 403 ASSET_MISMATCH. The shared key still works for PCs that have not enrolled, is refused (401 SHARED_KEY_REFUSED) for any PC that has, and can be switched off. A revoked credential gets 401 CREDENTIAL_INVALID so the agent re-enrolls. - The core collector accepts the credential for the computers, printers and endpoints collectors through two new plugin hooks exposed on shopdb.api (contract 0.33.0), and refuses the shared key for enrolled PCs. - Admin API: enrollment list, approve, reject, revoke (un-enroll), settings. Migration API: per-PC readiness with blockers, and upgrade waves that package the agent once and queue the existing agent upgrade per PC. - Migration 0023 creates endpointcredentials, endpointenrollrequests and endpointmigrationwaves, drops the certificate columns from assets where test databases have them, and renames app_fingerprint to appfingerprint. 0021 and 0022 keep their ids but no longer alter the core assets table. - Removed: client_cert.py, certificate renewal, the unauthenticated /test/client-cert route, nginx client certificate forwarding and the agent certificate script. The client-cert header could be forged on IIS, the key was readable by users, and the shared key remained a fallback. - GET /settings/server-cert returns the SPKI SHA-256 pin for GE-Enforce. Review fixes: - report_command_result used g without importing it; every result was a 500. - collect_log refuses setup answer files, appsettings, agent directories and key files at enqueue. - Fleet alerts: alerttimeout and alertfleet agree with the migration, the scope is validated and a malformed scope no longer reaches the whole fleet, and the ack race no longer raises NameError. - A refused break-glass override leaves no hold and no audit row. - Agent upgrade checks locks and permissions before repackaging, and writes the package atomically. - Server certificate upload validates the pair before replacing files; generation no longer builds an openssl config from user input; audit rows are actually written. - Health evaluation runs in a savepoint so a caught error cannot lose the check-in. - Required-app policies are found by appid, survive renames, and deleting an app deactivates policies that still name it. The UI and API now say such a policy only takes effect inside a business policy. - Backfill handles duplicate fingerprints and rolls back a failed batch. - Plugin settings routes skip a plugin whose hook raises. - The setup token is no longer printed to the log. - The naming gate checks top-level files again and handles odd file names. - Removed root scratch scripts, one of which carried a token. - Rewrote the fleet alert scheduling and app delta tests on real fixtures.
924 lines
39 KiB
Python
924 lines
39 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_a_refused_override_leaves_no_hold_and_no_audit_row(client, db):
|
|
"""Validation after the override branch must run before anything is written.
|
|
|
|
A targetless service action, and a second command while one is in flight,
|
|
are both refused. Neither may leave a policyoverridden audit row or a
|
|
PolicyActionHold behind, or the record claims a break-glass that never ran.
|
|
"""
|
|
from shopdb.api import AuditLog
|
|
from plugins.computers.models import PolicyActionHold
|
|
from plugins.endpoints.models import EndpointCommand
|
|
|
|
_enable_policies(db)
|
|
assetid = _endpoint(db, 'refusedbox')
|
|
_deny_everything(db, assetid)
|
|
headers = _user(db, client, 'refusedadmin',
|
|
['endpoints.view', 'endpoints.command'], admin=True)
|
|
|
|
missing_target = client.post(_url(assetid), headers=headers,
|
|
json={'action': 'service_status',
|
|
'policyoverride': True,
|
|
'overridereason': 'no target given'})
|
|
assert missing_target.status_code == 400, missing_target.get_json()
|
|
|
|
db.session.add(EndpointCommand(assetid=assetid, action='collect_now',
|
|
status=EndpointCommand.STATUS_PENDING))
|
|
db.session.commit()
|
|
in_flight = client.post(_url(assetid), headers=headers,
|
|
json={'action': 'service_status',
|
|
'target': 'spooler',
|
|
'policyoverride': True,
|
|
'overridereason': 'queue is busy'})
|
|
assert in_flight.status_code == 409, in_flight.get_json()
|
|
|
|
db.session.expire_all()
|
|
assert AuditLog.query.filter_by(action='policyoverridden').count() == 0
|
|
assert PolicyActionHold.query.filter_by(assetid=assetid).count() == 0
|
|
|
|
|
|
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']
|