ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
151d7c4f44089ecc941ca5661a92876c12e6182e
shopdb-flask/plugins/computers/api/routes.py
cproudlock 151d7c4f44 Anyone not signed in holds the guest role 
Whether somebody with no account may see something was being answered one
feature at a time: a securesend_allow_guests setting, an enable flag per kiosk,
and very nearly a third for guides - each with its own settings page, its own
vocabulary and its own hand-written branch inside a route. Every one of them
answered a permissions question, and this app already administers permissions
per role.

So a caller with no token is not nobody: they hold the seeded `guest` role.
require_permission consults it when there is no identity, which is the entire
mechanism. The role is seeded holding NOTHING, so an upgrade opens no door; a
site opens one by ticking a box in the same grid it uses for every other role,
and closes it the same way. Revocation is read on every request rather than
cached, because taking guest access away is what somebody does the moment they
realise what they opened, and it has to bite immediately.

The redundant @jwt_required() above require_permission is gone from 184 routes.
It was belt-and-braces before; now it is actively wrong, because it refuses a
request with no token before the permission check can have an opinion, so a
grant would appear to do nothing.

What a guest may hold is bounded. Any .view or .read permission is grantable.
A WRITE has to be blessed by whoever owns it, through the new
get_guest_grantable_permissions hook (contract 0.21.0), and Secure Send is the
only case that exists: sending a password to a vendor is the one write this
product means to offer somebody who never signs in. Everything else is refused
by the API and greyed out in the grid rather than warned about - a checkbox
that quietly makes deleting an asset anonymous is not a warning anybody reads
twice. A guest write records no creator: the trail says when and from where,
never who, and that is the reason for the bar.

GE-Enforce and Slides gained read permissions so they can be opened to guests
without handing out the manage right. Their read routes accept either through
require_any_permission, so nothing an existing manager could reach has moved.

The navigation endpoint now returns only what the caller can actually reach.
Without that a signed-out visitor gets a sidebar of the whole app where every
link bounces off the login page, which reads as a broken site rather than as a
closed door. A nav item may name several permissions; holding any one is
enough, which is what GE-Enforce and Slides need now that a viewer and a
manager hold different strings.

Two things this found on the way: current_user is a LocalProxy, so
`current_user is None` is False for a request with nobody behind it - harmless
under a required JWT, an AttributeError the moment verification became
optional. And an image pasted into a guide before its first save was never
attached to it, so it was visible to editors and invisible to every reader;
saving now attaches it, and `flask guides fix-images` repairs what is stored.

The guides guest setting written earlier today is gone. Granting guides.view to
the guest role replaces it, which is the point.
2026-08-27 12:51:58 -04:00

1173 lines
42 KiB
Python
Raw Blame History

"""Computers plugin API endpoints."""
from flask import Blueprint, request, Response, current_app
from flask_jwt_extended import jwt_required
from shopdb.api import db, Asset, AssetType, OperatingSystem, Application, AuditLog, Communication, CommunicationType, Setting, success_response, error_response, paginated_response, ErrorCodes, get_pagination_params, paginate_query
from ..models import Computer, ComputerType, ComputerInstalledApp, AccessProtocol, ComputerAccess
from shopdb.api import require_permission, require_role, apply_import_timestamps
computers_bp = Blueprint('computers', __name__)
# =============================================================================
# Computer Types
# =============================================================================
@computers_bp.route('/types', methods=['GET'])
@jwt_required(optional=True)
def list_computer_types():
"""List all computer types."""
page, per_page = get_pagination_params(request)
query = ComputerType.query
if request.args.get('active', 'true').lower() != 'false':
query = query.filter(ComputerType.isactive == True)
if search := request.args.get('search'):
query = query.filter(ComputerType.computertype.ilike(f'%{search}%'))
query = query.order_by(ComputerType.computertype)
items, total = paginate_query(query, page, per_page)
data = [t.to_dict() for t in items]
return paginated_response(data, page, per_page, total)
@computers_bp.route('/types/<int:type_id>', methods=['GET'])
@jwt_required(optional=True)
def get_computer_type(type_id: int):
"""Get a single computer type."""
t = db.session.get(ComputerType, type_id)
if not t:
return error_response(
ErrorCodes.NOT_FOUND,
f'Computer type with ID {type_id} not found',
http_code=404
)
return success_response(t.to_dict())
@computers_bp.route('/types', methods=['POST'])
@require_permission('computers.create')
def create_computer_type():
"""Create a new computer type."""
data = request.get_json()
if not data or not data.get('computertype'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'computertype is required')
existing = ComputerType.query.filter_by(computertype=data['computertype']).first()
if existing:
if not existing.isactive:
# Adding a name that matches a deactivated type revives it.
existing.isactive = True
for key in ('description', 'icon', 'color'):
if data.get(key) is not None:
setattr(existing, key, data[key])
db.session.commit()
return success_response(existing.to_dict(), message='Reactivated existing computer type')
return error_response(
ErrorCodes.CONFLICT,
f"Computer type '{data['computertype']}' already exists",
http_code=409
)
t = ComputerType(
computertype=data['computertype'],
description=data.get('description'),
icon=data.get('icon'), color=data.get('color')
)
db.session.add(t)
db.session.commit()
return success_response(t.to_dict(), message='Computer type created', http_code=201)
@computers_bp.route('/types/<int:type_id>', methods=['PUT'])
@require_permission('computers.edit')
def update_computer_type(type_id: int):
"""Update a computer type."""
t = db.session.get(ComputerType, type_id)
if not t:
return error_response(
ErrorCodes.NOT_FOUND,
f'Computer type with ID {type_id} not found',
http_code=404
)
data = request.get_json()
if not data:
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
if 'computertype' in data and data['computertype'] != t.computertype:
if ComputerType.query.filter_by(computertype=data['computertype']).first():
return error_response(
ErrorCodes.CONFLICT,
f"Computer type '{data['computertype']}' already exists",
http_code=409
)
for key in ['computertype', 'description', 'icon', 'color', 'isactive']:
if key in data:
setattr(t, key, data[key])
db.session.commit()
return success_response(t.to_dict(), message='Computer type updated')
@computers_bp.route('/types/<int:type_id>', methods=['DELETE'])
@require_permission('computers.delete')
def delete_computer_type(type_id: int):
"""Delete a computer type. Refused if any PC still uses it."""
t = db.session.get(ComputerType, type_id)
if not t:
return error_response(ErrorCodes.NOT_FOUND, 'Computer type not found', http_code=404)
inuse = Computer.query.filter_by(computertypeid=type_id).count()
if inuse:
return error_response(ErrorCodes.CONFLICT,
f"Cannot delete: {inuse} PC(s) still use this type", http_code=409)
db.session.delete(t)
db.session.commit()
return success_response(message='Computer type deleted')
# =============================================================================
# Access protocol catalog (VNC / WinRM / RDP / ...) - admin-managed
# =============================================================================
@computers_bp.route('/protocols', methods=['GET'])
@jwt_required(optional=True)
def list_protocols():
"""List access protocols. ?active=false includes disabled ones."""
query = AccessProtocol.query
if request.args.get('active', 'true').lower() != 'false':
query = query.filter(AccessProtocol.isactive == True)
protocols = query.order_by(AccessProtocol.name).all()
return success_response([p.to_dict() for p in protocols])
@computers_bp.route('/protocols', methods=['POST'])
@require_permission('computers.edit')
def create_protocol():
data = request.get_json() or {}
if not (data.get('name') and data.get('scheme') and data.get('linktemplate')):
return error_response(ErrorCodes.VALIDATION_ERROR, 'name, scheme and linktemplate are required')
if AccessProtocol.query.filter_by(name=data['name']).first():
return error_response(ErrorCodes.CONFLICT, f"Protocol '{data['name']}' already exists", http_code=409)
p = AccessProtocol(
name=data['name'],
scheme=data['scheme'],
defaultport=data.get('defaultport') or None,
linktemplate=data['linktemplate'],
isactive=data.get('isactive', True),
)
db.session.add(p)
db.session.commit()
return success_response(p.to_dict(), message='Protocol created', http_code=201)
@computers_bp.route('/protocols/<int:protocol_id>', methods=['PUT', 'PATCH'])
@require_permission('computers.edit')
def update_protocol(protocol_id):
p = db.session.get(AccessProtocol, protocol_id)
if not p:
return error_response(ErrorCodes.NOT_FOUND, 'Protocol not found', http_code=404)
data = request.get_json() or {}
for field in ('name', 'scheme', 'linktemplate'):
if data.get(field):
setattr(p, field, data[field])
if 'defaultport' in data:
p.defaultport = data['defaultport'] or None
if 'isactive' in data:
p.isactive = bool(data['isactive'])
db.session.commit()
return success_response(p.to_dict(), message='Protocol updated')
@computers_bp.route('/protocols/<int:protocol_id>', methods=['DELETE'])
@require_permission('computers.edit')
def delete_protocol(protocol_id):
p = db.session.get(AccessProtocol, protocol_id)
if not p:
return error_response(ErrorCodes.NOT_FOUND, 'Protocol not found', http_code=404)
# If any PC still references it, deactivate rather than hard-delete.
if ComputerAccess.query.filter_by(protocolid=protocol_id).first():
p.isactive = False
db.session.commit()
return success_response(message='Protocol is in use; deactivated instead of deleted')
db.session.delete(p)
db.session.commit()
return success_response(message='Protocol deleted')
def _pc_access_domain():
# Contract-pure read of the pc_access_domain setting (no core.api import).
row = Setting.query.filter_by(key='pc_access_domain').first()
return ((row.value if row else '') or '').strip()
def _computer_access_links(comp, domain=None):
"""Resolved remote-access links for a computer: each enabled protocol's
template filled with the PC hostname joined to the pc_access_domain setting.
A hostname that is already an FQDN (has a dot) is used as-is. Pass domain
when calling in a loop to avoid one settings lookup per computer."""
if domain is None:
domain = _pc_access_domain()
hostname = (comp.hostname or '').strip()
if not hostname:
host = ''
elif '.' in hostname or not domain:
host = hostname
else:
host = f"{hostname}.{domain}"
links = []
for am in comp.accessmethods:
protocol = am.protocol
if not (am.isactive and protocol and protocol.isactive):
continue
port = am.portoverride or protocol.defaultport
link = None
if host:
try:
link = protocol.linktemplate.format(
host=host,
port=(port if port is not None else ''),
scheme=protocol.scheme,
)
except (KeyError, IndexError, ValueError):
link = None
links.append({
'id': am.id,
'protocolid': protocol.protocolid,
'name': protocol.name,
'scheme': protocol.scheme,
'port': port,
'portoverride': am.portoverride,
'link': link,
})
return links
def _sync_access_methods(comp, data):
"""Replace a computer's enabled protocols from data['accessmethods'] (a list
of {protocolid, portoverride?}). No-op if the key is absent, so callers that
don't touch access aren't affected."""
if 'accessmethods' not in data:
return
desired = data.get('accessmethods') or []
ComputerAccess.query.filter_by(computerid=comp.computerid).delete()
seen = set()
for m in desired:
try:
pid = int(m.get('protocolid'))
except (TypeError, ValueError):
continue
if pid in seen:
continue
seen.add(pid)
port = m.get('portoverride')
try:
port = int(port) if port not in (None, '') else None
except (TypeError, ValueError):
port = None
db.session.add(ComputerAccess(
computerid=comp.computerid,
protocolid=pid,
portoverride=port,
isactive=True,
))
# =============================================================================
# Computers CRUD
# =============================================================================
@computers_bp.route('/display-kiosks', methods=['GET'])
@jwt_required(optional=True)
def list_display_kiosks():
"""Reporting display kiosks, for the Dashboard Defaults picker.
Returns the computers whose type is the one gea-shopfloor-display maps to
(default 'Kiosk'), each with its DERIVED FQDN (F<serial>.<domain>, domain
from the display_fqdn_domain setting) so an admin picks a kiosk from a
dropdown instead of typing an IP/FQDN. Keeps the same F<serial>.<domain>
format as core derive_display_fqdn (duplicated to avoid a contract bump).
"""
from ..pctypemap import pctype_mapping
from shopdb.api import Setting
display_type_name = pctype_mapping().get('gea-shopfloor-display', 'Kiosk')
ctype = ComputerType.query.filter_by(computertype=display_type_name).first()
if not ctype:
return success_response([])
domain = (Setting.get('display_fqdn_domain', 'device.geaerospace.net') # ADR-015-OK: GE Aerospace-wide domain, and only the DEFAULT of a documented setting every site can override.
or 'device.geaerospace.net').strip().strip('.') # ADR-015-OK: GE Aerospace-wide domain, and only the DEFAULT of a documented setting every site can override.
rows = (db.session.query(Computer).join(Asset)
.filter(Computer.computertypeid == ctype.computertypeid,
Asset.isactive == True)
.order_by(Computer.hostname).all())
out = []
for comp in rows:
serial = (comp.asset.serialnumber or '').strip() if comp.asset else ''
out.append({
'computerid': comp.computerid,
'hostname': comp.hostname,
'serialnumber': serial or None,
'fqdn': f'F{serial}.{domain}'.lower() if serial else None,
})
return success_response(out)
@computers_bp.route('', methods=['GET'])
@jwt_required(optional=True)
def list_computers():
"""
List all computers with filtering and pagination.
Query parameters:
- page, per_page: Pagination
- active: Filter by active status
- search: Search by asset number, name, or hostname
- type_id: Filter by computer type ID
- os_id: Filter by operating system ID
- location_id: Filter by location ID
- businessunit_id: Filter by business unit ID
- shopfloor: Filter by shopfloor flag (true/false)
"""
page, per_page = get_pagination_params(request)
# Join Computer with Asset
query = db.session.query(Computer).join(Asset)
# Active filter
if request.args.get('active', 'true').lower() != 'false':
query = query.filter(Asset.isactive == True)
# Exact-match natural-key lookup for idempotent import (asset number).
if exactassetnumber := request.args.get('assetnumber'):
query = query.filter(Asset.assetnumber == exactassetnumber)
# Search filter. Covers the type name too - it is a column in the list, so
# searching 'Standard' must find the PCs of that type. Outer join so a PC
# with no type still matches on its own fields.
if search := request.args.get('search'):
pattern = f'%{search}%'
query = query.outerjoin(
ComputerType, Computer.computertypeid == ComputerType.computertypeid
).filter(
db.or_(
Asset.assetnumber.ilike(pattern),
Asset.name.ilike(pattern),
Asset.serialnumber.ilike(pattern),
# The optional identifiers too (ADR-001). Global search matches
# these, and a tag read off the machine has to find it here as
# well - this box is where someone holding the label looks.
Asset.gaugelabreference.ilike(pattern),
Asset.maintenancereference.ilike(pattern),
Computer.hostname.ilike(pattern),
ComputerType.computertype.ilike(pattern)
)
)
# Computer type filter
if type_id := request.args.get('typeid', request.args.get('type_id')):
query = query.filter(Computer.computertypeid == int(type_id))
# OS filter
if os_id := request.args.get('osid', request.args.get('os_id')):
query = query.filter(Computer.osid == int(os_id))
# Location filter
if location_id := request.args.get('locationid', request.args.get('location_id')):
query = query.filter(Asset.locationid == int(location_id))
# Business unit filter
if bu_id := request.args.get('businessunitid', request.args.get('businessunit_id')):
query = query.filter(Asset.businessunitid == int(bu_id))
# Shopfloor filter (by the Shopfloor computer type)
if shopfloor := request.args.get('shopfloor'):
sf = ComputerType.query.filter_by(computertype='Shopfloor').first()
sf_id = sf.computertypeid if sf else -1
if shopfloor.lower() == 'true':
query = query.filter(Computer.computertypeid == sf_id)
else:
query = query.filter(db.or_(Computer.computertypeid != sf_id,
Computer.computertypeid.is_(None)))
# Sorting
sort_by = request.args.get('sort', 'hostname')
sort_dir = request.args.get('dir', 'asc')
if sort_by == 'hostname':
col = Computer.hostname
elif sort_by == 'assetnumber':
col = Asset.assetnumber
elif sort_by == 'name':
col = Asset.name
elif sort_by == 'lastreporteddate':
col = Computer.lastreporteddate
else:
col = Computer.hostname
query = query.order_by(col.desc() if sort_dir == 'desc' else col)
items, total = paginate_query(query, page, per_page)
# Build response with both asset and computer data
data = []
accessdomain = _pc_access_domain()
for comp in items:
item = comp.asset.to_dict() if comp.asset else {}
item['computer'] = comp.to_dict()
item['accessmethods'] = _computer_access_links(comp, domain=accessdomain)
data.append(item)
return paginated_response(data, page, per_page, total)
@computers_bp.route('/<int:computer_id>', methods=['GET'])
@jwt_required(optional=True)
def get_computer(computer_id: int):
"""Get a single computer with full details."""
comp = db.session.get(Computer, computer_id)
if not comp:
return error_response(
ErrorCodes.NOT_FOUND,
f'Computer with ID {computer_id} not found',
http_code=404
)
result = comp.asset.to_dict() if comp.asset else {}
result['computer'] = comp.to_dict()
result['communications'] = [
c.to_dict() for c in
Communication.query.filter_by(assetid=comp.assetid).all()
]
result['accessmethods'] = _computer_access_links(comp)
return success_response(result)
@computers_bp.route('/by-asset/<int:asset_id>', methods=['GET'])
@jwt_required(optional=True)
def get_computer_by_asset(asset_id: int):
"""Get computer data by asset ID."""
comp = Computer.query.filter_by(assetid=asset_id).first()
if not comp:
return error_response(
ErrorCodes.NOT_FOUND,
f'Computer for asset {asset_id} not found',
http_code=404
)
result = comp.asset.to_dict() if comp.asset else {}
result['computer'] = comp.to_dict()
return success_response(result)
@computers_bp.route('/by-hostname/<hostname>', methods=['GET'])
@jwt_required(optional=True)
def get_computer_by_hostname(hostname: str):
"""Get computer by hostname."""
comp = Computer.query.filter_by(hostname=hostname).first()
if not comp:
return error_response(
ErrorCodes.NOT_FOUND,
f'Computer with hostname {hostname} not found',
http_code=404
)
result = comp.asset.to_dict() if comp.asset else {}
result['computer'] = comp.to_dict()
return success_response(result)
@computers_bp.route('', methods=['POST'])
@require_permission('computers.create')
def create_computer():
"""
Create new computer (creates both Asset and Computer records).
Required fields:
- assetnumber: Business identifier
Optional fields:
- name, serialnumber, statusid, locationid, businessunitid
- computertypeid, hostname, osid
- mapx, mapy, notes
"""
data = request.get_json()
if not data:
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
if not data.get('assetnumber'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'assetnumber is required')
# Check for duplicate assetnumber
if Asset.query.filter_by(assetnumber=data['assetnumber']).first():
return error_response(
ErrorCodes.CONFLICT,
f"Asset with number '{data['assetnumber']}' already exists",
http_code=409
)
# Check for duplicate hostname
if data.get('hostname'):
if Computer.query.filter_by(hostname=data['hostname']).first():
return error_response(
ErrorCodes.CONFLICT,
f"Computer with hostname '{data['hostname']}' already exists",
http_code=409
)
# Get computer asset type
computer_type = AssetType.query.filter_by(assettype='computer').first()
if not computer_type:
return error_response(
ErrorCodes.INTERNAL_ERROR,
'Computer asset type not found. Plugin may not be properly installed.',
http_code=500
)
# Create the core asset
asset = Asset(
assetnumber=data['assetnumber'],
name=data.get('name'),
serialnumber=data.get('serialnumber'),
gaugelabreference=data.get('gaugelabreference'),
maintenancereference=data.get('maintenancereference'),
assettypeid=computer_type.assettypeid,
statusid=data.get('statusid', 1),
locationid=data.get('locationid'),
businessunitid=data.get('businessunitid'),
mapx=data.get('mapx'),
levelid=data.get('levelid'),
mapy=data.get('mapy'),
notes=data.get('notes')
)
db.session.add(asset)
db.session.flush() # Get the assetid
# Create the computer extension
comp = Computer(
assetid=asset.assetid,
computertypeid=data.get('computertypeid'),
hostname=data.get('hostname'),
osid=data.get('osid'),
vendorid=data.get('vendorid'),
modelnumberid=data.get('modelnumberid'),
loggedinuser=data.get('loggedinuser'),
lastreporteddate=data.get('lastreporteddate'),
lastboottime=data.get('lastboottime')
)
db.session.add(comp)
db.session.flush()
# Optional primary IP communication
if data.get('ipaddress'):
ip_comtype = CommunicationType.query.filter_by(comtype='IP').first()
if ip_comtype:
db.session.add(Communication(
assetid=asset.assetid,
comtypeid=ip_comtype.comtypeid,
ipaddress=data['ipaddress'],
isprimary=True,
))
# Remote-access protocols
_sync_access_methods(comp, data)
# Preserve legacy timestamps in import mode (no-op otherwise)
apply_import_timestamps(asset, data)
# Audit log
AuditLog.log('created', 'Computer', entityid=comp.computerid,
entityname=data.get('hostname') or data['assetnumber'])
db.session.commit()
result = asset.to_dict()
result['computer'] = comp.to_dict()
result['accessmethods'] = _computer_access_links(comp)
return success_response(result, message='Computer created', http_code=201)
@computers_bp.route('/<int:computer_id>', methods=['PUT'])
@require_permission('computers.edit')
def update_computer(computer_id: int):
"""Update computer (both Asset and Computer records)."""
comp = db.session.get(Computer, computer_id)
if not comp:
return error_response(
ErrorCodes.NOT_FOUND,
f'Computer with ID {computer_id} not found',
http_code=404
)
data = request.get_json()
if not data:
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
asset = comp.asset
# An UPDATE may not blank the asset number. Create validates it and the
# column is NOT NULL, but the conflict check below only runs when the value
# DIFFERS, and '' never collides with anything - so a payload carrying an
# empty assetnumber wrote it straight through. A form that loaded blank
# (a failed reference load, a partial fetch) then saved the blank over a
# real record.
if 'assetnumber' in data and not (data['assetnumber'] or '').strip():
return error_response(
ErrorCodes.VALIDATION_ERROR,
'assetnumber cannot be empty')
# Check for conflicting assetnumber
if 'assetnumber' in data and data['assetnumber'] != asset.assetnumber:
if Asset.query.filter_by(assetnumber=data['assetnumber']).first():
return error_response(
ErrorCodes.CONFLICT,
f"Asset with number '{data['assetnumber']}' already exists",
http_code=409
)
# Check for conflicting hostname
if 'hostname' in data and data['hostname'] != comp.hostname:
existing = Computer.query.filter_by(hostname=data['hostname']).first()
if existing and existing.computerid != computer_id:
return error_response(
ErrorCodes.CONFLICT,
f"Computer with hostname '{data['hostname']}' already exists",
http_code=409
)
# Track changes for audit log
changes = {}
# Update asset fields
asset_fields = ['assetnumber', 'name', 'serialnumber', 'gaugelabreference',
'maintenancereference', 'statusid',
'locationid', 'businessunitid', 'mapx', 'mapy', 'levelid', 'notes', 'isactive']
for key in asset_fields:
if key in data:
old_val = getattr(asset, key)
new_val = data[key]
if old_val != new_val:
changes[key] = {'old': old_val, 'new': new_val}
setattr(asset, key, data[key])
# Update computer fields
computer_fields = ['computertypeid', 'hostname', 'osid', 'vendorid',
'modelnumberid', 'loggedinuser', 'lastreporteddate',
'lastboottime']
for key in computer_fields:
if key in data:
old_val = getattr(comp, key)
new_val = data[key]
if old_val != new_val:
changes[key] = {'old': old_val, 'new': new_val}
setattr(comp, key, data[key])
# Upsert the primary IP communication so a single PUT covers it
if 'ipaddress' in data:
ip = (data.get('ipaddress') or '').strip()
primary = Communication.query.filter_by(
assetid=asset.assetid, isprimary=True).first()
if ip:
if primary:
primary.ipaddress = ip
else:
ip_comtype = CommunicationType.query.filter_by(comtype='IP').first()
if ip_comtype:
db.session.add(Communication(
assetid=asset.assetid, comtypeid=ip_comtype.comtypeid,
ipaddress=ip, isprimary=True))
elif primary:
primary.ipaddress = None
# Remote-access protocols
_sync_access_methods(comp, data)
# Audit log if there were changes
if changes:
AuditLog.log('updated', 'Computer', entityid=comp.computerid,
entityname=comp.hostname or asset.assetnumber, changes=changes)
apply_import_timestamps(asset, data)
db.session.commit()
result = asset.to_dict()
result['computer'] = comp.to_dict()
result['accessmethods'] = _computer_access_links(comp)
return success_response(result, message='Computer updated')
@computers_bp.route('/<int:computer_id>', methods=['DELETE'])
@require_permission('computers.delete')
def delete_computer(computer_id: int):
"""Delete (soft delete) computer."""
comp = db.session.get(Computer, computer_id)
if not comp:
return error_response(
ErrorCodes.NOT_FOUND,
f'Computer with ID {computer_id} not found',
http_code=404
)
# Soft delete the asset
comp.asset.isactive = False
# Audit log
AuditLog.log('deleted', 'Computer', entityid=comp.computerid,
entityname=comp.hostname or comp.asset.assetnumber)
db.session.commit()
return success_response(message='Computer deleted')
# =============================================================================
# Installed Applications
# =============================================================================
@computers_bp.route('/<int:computer_id>/apps', methods=['GET'])
@jwt_required(optional=True)
def get_installed_apps(computer_id: int):
"""Get all installed applications for a computer."""
comp = db.session.get(Computer, computer_id)
if not comp:
return error_response(
ErrorCodes.NOT_FOUND,
f'Computer with ID {computer_id} not found',
http_code=404
)
apps = ComputerInstalledApp.query.filter_by(
computerid=computer_id,
isactive=True
).all()
data = [app.to_dict() for app in apps]
return success_response(data)
@computers_bp.route('/<int:computer_id>/apps', methods=['POST'])
@require_permission('computers.create')
def add_installed_app(computer_id: int):
"""Add an installed application to a computer."""
comp = db.session.get(Computer, computer_id)
if not comp:
return error_response(
ErrorCodes.NOT_FOUND,
f'Computer with ID {computer_id} not found',
http_code=404
)
data = request.get_json()
if not data or not data.get('appid'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'appid is required')
appid = data['appid']
# Validate app exists
if not db.session.get(Application, appid):
return error_response(ErrorCodes.NOT_FOUND, f'Application {appid} not found', http_code=404)
# Check for duplicate
existing = ComputerInstalledApp.query.filter_by(
computerid=computer_id,
appid=appid
).first()
if existing:
if existing.isactive:
return error_response(
ErrorCodes.CONFLICT,
'This application is already installed on this computer',
http_code=409
)
else:
# Reactivate
existing.isactive = True
existing.appversionid = data.get('appversionid')
db.session.commit()
return success_response(existing.to_dict(), message='Application reinstalled')
# Create new installation record
installed = ComputerInstalledApp(
computerid=computer_id,
appid=appid,
appversionid=data.get('appversionid')
)
db.session.add(installed)
db.session.commit()
return success_response(installed.to_dict(), message='Application installed', http_code=201)
@computers_bp.route('/<int:computer_id>/apps/<int:app_id>', methods=['DELETE'])
@require_permission('computers.delete')
def remove_installed_app(computer_id: int, app_id: int):
"""Remove an installed application from a computer."""
installed = ComputerInstalledApp.query.filter_by(
computerid=computer_id,
appid=app_id,
isactive=True
).first()
if not installed:
return error_response(
ErrorCodes.NOT_FOUND,
'Installation record not found',
http_code=404
)
installed.isactive = False
db.session.commit()
return success_response(message='Application uninstalled')
# =============================================================================
# Status Reporting
# =============================================================================
@computers_bp.route('/<int:computer_id>/report', methods=['POST'])
@require_permission('computers.create')
def report_status(computer_id: int):
"""
Report computer status (for agent-based reporting).
This endpoint can be called periodically by a client agent
to update status information.
"""
comp = db.session.get(Computer, computer_id)
if not comp:
return error_response(
ErrorCodes.NOT_FOUND,
f'Computer with ID {computer_id} not found',
http_code=404
)
data = request.get_json() or {}
# Update status fields
from datetime import datetime, timezone
comp.lastreporteddate = datetime.now(timezone.utc).replace(tzinfo=None)
if 'loggedinuser' in data:
comp.loggedinuser = data['loggedinuser']
if 'lastboottime' in data:
comp.lastboottime = data['lastboottime']
db.session.commit()
return success_response(message='Status reported')
# =============================================================================
# Dashboard
# =============================================================================
@computers_bp.route('/dashboard/summary', methods=['GET'])
@jwt_required(optional=True)
def dashboard_summary():
"""Get computer dashboard summary data."""
# Total active computers
total = db.session.query(Computer).join(Asset).filter(
Asset.isactive == True
).count()
# Count by computer type
by_type = db.session.query(
ComputerType.computertype,
db.func.count(Computer.computerid)
).join(Computer, Computer.computertypeid == ComputerType.computertypeid
).join(Asset, Asset.assetid == Computer.assetid
).filter(Asset.isactive == True
).group_by(ComputerType.computertype
).all()
# Count by OS
by_os = db.session.query(
OperatingSystem.osname,
db.func.count(Computer.computerid)
).join(Computer, Computer.osid == OperatingSystem.osid
).join(Asset, Asset.assetid == Computer.assetid
).filter(Asset.isactive == True
).group_by(OperatingSystem.osname
).all()
# Count shopfloor vs non-shopfloor
sf = ComputerType.query.filter_by(computertype='Shopfloor').first()
shopfloor_count = db.session.query(Computer).join(Asset).filter(
Asset.isactive == True,
Computer.computertypeid == (sf.computertypeid if sf else -1)
).count()
return success_response({
'total': total,
'bytype': [{'type': t, 'count': c} for t, c in by_type],
'byos': [{'os': o, 'count': c} for o, c in by_os],
'shopfloor': shopfloor_count,
'nonshopfloor': total - shopfloor_count
})
@computers_bp.route('/dashboard/quiet', methods=['GET'])
@require_permission('computers.view')
def dashboard_quiet():
"""PCs that WERE reporting, have stopped, and are still meant to be in use.
Three filters, and each removes a population that would otherwise be noise:
Never reported at all is EXCLUDED. Those are usually records created by
hand or imported, not bays that broke - listing them buries the PCs that
actually changed state behind rows nobody is going to act on.
Not In Use is EXCLUDED. A PC in Repair, Inventory or Retired is silent on
purpose; that is the status doing its job, not a fault.
Soft-deleted is excluded for the same reason.
What is left is the real signal: a machine that was working, is not now, and
nobody has marked as anything else. Silence is the only evidence available -
there is no heartbeat separate from the report.
"""
from datetime import datetime, timedelta, timezone
from shopdb.api import AssetStatus
hours = 24
setting = Setting.query.filter_by(key='computers_quietreporthours').first()
if setting and (setting.value or '').strip():
try:
hours = int(setting.value)
except (TypeError, ValueError):
pass
cutoff = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(hours=hours)
now = datetime.now(timezone.utc).replace(tzinfo=None)
rows = (db.session.query(Computer, Asset)
.join(Asset, Asset.assetid == Computer.assetid)
.join(AssetStatus, AssetStatus.statusid == Asset.statusid)
.filter(Asset.isactive.is_(True),
AssetStatus.status == 'In Use',
Computer.lastreporteddate.isnot(None),
Computer.lastreporteddate < cutoff)
.all())
out = [{
'computerid': comp.computerid,
'hostname': comp.hostname,
'lastreported': comp.lastreporteddate.isoformat() + 'Z',
'quietdays': (now - comp.lastreporteddate).days,
} for comp, _asset in rows]
out.sort(key=lambda r: -r['quietdays'])
return success_response(out[:50])
@computers_bp.route('/dashboard/sharedmachines', methods=['GET'])
@require_permission('computers.view')
def dashboard_sharedmachines():
"""Machine numbers claimed by more than one PC, with nothing filed under
them.
Several devices genuinely sharing a number is legitimate - part markers do
it - and those are modelled: each device is its own asset filed `partof` the
operation, so the operation has CHILD ASSETS. Two PCs carrying the same
machine number by mistake looks identical from a count and has none. That
distinction is the whole card; without it this would list correct data
alongside faults and get ignored.
Promoted from `flask relationships check-shared-machines`, which answers the
same question and which nobody will remember to run. This one found seven
mis-numbered bays that had been that way for weeks.
"""
from shopdb.api import AssetRelationship, RelationshipType
from sqlalchemy.orm import aliased
controls = RelationshipType.query.filter_by(
relationshiptype='controls').first()
partof = RelationshipType.query.filter_by(relationshiptype='partof').first()
if not controls:
return success_response([])
pcasset = aliased(Asset)
machineasset = aliased(Asset)
rows = (db.session.query(machineasset.assetid, machineasset.assetnumber,
pcasset.assetnumber)
.select_from(AssetRelationship)
.join(pcasset, AssetRelationship.sourceassetid == pcasset.assetid)
.join(machineasset,
AssetRelationship.targetassetid == machineasset.assetid)
.filter(AssetRelationship.relationshiptypeid ==
controls.relationshiptypeid,
AssetRelationship.label == 'collector:machine',
AssetRelationship.isactive.is_(True))
.all())
bymachine = {}
for assetid, machinenumber, pcnumber in rows:
bymachine.setdefault((assetid, machinenumber), []).append(pcnumber)
out = []
for (assetid, machinenumber), pcs in bymachine.items():
if len(pcs) < 2:
continue
children = 0
if partof:
children = (AssetRelationship.query
.filter_by(targetassetid=assetid,
relationshiptypeid=partof.relationshiptypeid,
isactive=True)
.count())
if children:
continue # modelled: the devices are assets in their own right
out.append({
'assetid': assetid,
'machinenumber': machinenumber,
'pccount': len(pcs),
'pcs': ', '.join(sorted(p for p in pcs if p)),
})
out.sort(key=lambda r: -r['pccount'])
return success_response(out)
# =============================================================================
# Collector client script
# =============================================================================
CLIENT_SCRIPT_NAME = 'Report-AssetToShopDB.ps1'
def _client_script_path():
"""The reporter shipped with this plugin, which is the single source."""
import os
return os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
'client', CLIENT_SCRIPT_NAME)
def _setting_value(key):
row = Setting.query.filter_by(key=key).first()
return ((row.value if row else '') or '').strip()
def _generate_client_script(source: str, baseurl: str, ranges: str,
version: str, generatedon: str) -> str:
"""Stamp a site's own values into the reporter's parameter defaults.
ONLY the defaults are substituted, never the body: the file in the repo
stays runnable as-is, so there is no second copy to drift. Everything
stamped here is overridable at runtime - the parameter still wins, then the
registry - because a bay may need to differ from its site.
The collector key is NOT stamped in. This file lands on every shop-floor PC,
and a token in a file on hundreds of bays cannot be rotated quietly; it is
read from HKLM:\\SOFTWARE\\GE\\ShopDB CollectorKey, provisioned per
ADOPTING-AT-ANOTHER-SITE.md.
"""
apiurl = baseurl.rstrip('/') + '/api/collector/computers' if baseurl else ''
header = (
'# GENERATED by ShopDB {version} on {generatedon}\n'
'# for {baseurl}\n'
'#\n'
'# Re-download after upgrading ShopDB: this copy matches that server\'s\n'
'# collector contract. Edits here are lost on the next download - change\n'
'# the site settings instead, or pass -ApiUrl / -AllowedRanges.\n'
'#\n'
'# The collector key is deliberately NOT in this file. Provision it as\n'
'# HKLM:\\SOFTWARE\\GE\\ShopDB CollectorKey - see the adoption guide.\n'
'\n'
).format(version=version, generatedon=generatedon,
baseurl=baseurl or 'an unconfigured site (set site_base_url)')
out = source
if apiurl:
old = "[string]$ApiUrl = ''"
assert old in out, 'the reporter no longer declares $ApiUrl as expected'
out = out.replace(old, "[string]$ApiUrl = '{0}'".format(apiurl), 1)
if ranges:
old = "[string]$AllowedRanges = ''"
assert old in out, 'the reporter no longer declares $AllowedRanges as expected'
out = out.replace(old, "[string]$AllowedRanges = '{0}'".format(ranges), 1)
return header + out
@computers_bp.route('/client-script', methods=['GET'])
@jwt_required()
@require_role('admin')
def download_client_script():
"""The collector reporter, stamped with THIS site's values.
Admin-only. It carries no secret, but it does state a site's URL and its
internal ranges, which is configuration rather than something to hand out.
"""
import datetime
import hashlib
import os
path = _client_script_path()
if not os.path.isfile(path):
return error_response(ErrorCodes.NOT_FOUND,
'The collector script is not present in this install',
http_code=404)
with open(path, 'r', encoding='utf-8') as handle:
source = handle.read()
# A site that has not set its public URL still gets a usable script: the
# browsing origin is the server the admin is talking to right now.
baseurl = _setting_value('site_base_url') or request.url_root
# From config, not an import: a plugin reaching into core is an ADR-002
# violation and the contract test fails the build for it.
version = current_app.config.get('VERSION') or 'unknown'
generatedon = datetime.datetime.now(datetime.timezone.utc).strftime('%Y-%m-%d')
try:
body = _generate_client_script(
source, baseurl.strip(), _setting_value('computers_routableranges'),
version, generatedon)
except AssertionError as exc:
return error_response(ErrorCodes.INTERNAL_ERROR, str(exc), http_code=500)
digest = hashlib.sha256(body.encode('utf-8')).hexdigest()
return Response(
body,
mimetype='text/plain; charset=utf-8',
headers={
'Content-Disposition': 'attachment; filename={0}'.format(CLIENT_SCRIPT_NAME),
# Published so a deployment can verify what it fetched, the same way
# the installer publishes one.
'X-Script-Sha256': digest,
})
Reference in New Issue View Git Blame Copy Permalink