diff --git a/.gitignore b/.gitignore index 75d4de9..f69038d 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,8 @@ settings.ini +.env +/env +/__pychache__ +.DS_Store +__pycache__/snipe.cpython-313.pyc + +__pycache__/mosyle.cpython-313.pyc diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..1ba8aff --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,312 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## Project Overview + +MosyleSnipeSync is a Python-based synchronization tool that integrates Apple device data between two mobile device management platforms: +- **Mosyle**: Apple-specific device management system +- **Snipe-IT**: Open-source asset management and IT inventory system + +The tool pulls device information from Mosyle (Mac, iOS, tvOS), creates/updates assets in Snipe-IT, manages device assignments, and syncs asset tags between systems. + +## Architecture + +### Core Components + +**Main Entry Point** (`main.py:1-135`) +- Orchestrates the sync workflow for each device type (mac, ios, tvos) +- Loads configuration from `settings.ini` +- For each device type: fetches devices from Mosyle → processes each device → updates Snipe-IT +- Uses `rich` library for progress bar visualization + +**Mosyle Integration** (`mosyle.py`) +- `Mosyle` class handles authentication and API communication with Mosyle +- Login flow: uses `access_token`, `email`, and `password` to obtain JWT token via `/login` endpoint +- Key methods: + - `list(os, page)`: Paginated device listing for a specific OS type + - `setAssetTag(serialnumber, tag)`: Updates device asset tag in Mosyle +- Maintains authenticated session with Bearer token in headers + +**Snipe-IT Integration** (`snipe.py`) +- `Snipe` class manages all Snipe-IT API interactions +- Constructor takes extensive config: API key, categories, fieldsets, rate limit, manufacturer ID +- Key responsibilities: + - **Model Management**: Creates/searches Mac, iOS, tvOS models with hardcoded custom field IDs + - **Asset Operations**: Creates, updates, lists hardware; manages assignments/checkouts + - **Image Handling**: Fetches Apple device images from `api.appledb.dev` and encodes as base64 data URIs + - **Rate Limiting**: Tracks requests and sleeps for 60s when limit (default 120/min) is exceeded + - **Retry Logic**: Implements exponential backoff for 429/5xx errors up to 5 attempts +- Payload building maps Mosyle device fields to hardcoded Snipe-IT custom field IDs (e.g., `_snipeit_cpu_family_7`) + +**Apple Info Utility** (`appleInfo.py`) +- Standalone script for retroactively updating images on Apple models in Snipe-IT +- Processes all models, filters by Apple manufacturer, downloads images via AppleDB if missing +- Useful for backfill operations + +### Data Flow + +1. Configuration parameters (including Mosyle credentials) loaded from `settings.ini` +2. For each device type in `settings.ini` → `mosyle.deviceTypes`: + - Query Mosyle for devices (paginated if using "all" calltype) + - For each Mosyle device: + - Search Snipe-IT by serial number + - If model doesn't exist, create it (with image from AppleDB if `apple_image_check` enabled) + - Create or update asset with device data + - Sync user assignment if device has console-managed user + - Update Mosyle asset tag with Snipe-IT's generated asset tag + +### Configuration + +**settings.ini** has 3 sections: +- `[mosyle]`: Mosyle API credentials and device type filtering +- `[snipe-it]`: Snipe-IT URL, API key, category/fieldset IDs, rate limit, image checking +- `[api-mapping]`: Field mapping (partially defined, expandable) + +Critical custom field IDs in `snipe.py:buildPayloadFromMosyle()` are hardcoded: +- `_snipeit_cpu_family_7`, `_snipeit_percent_disk_5`, `_snipeit_available_disk_5`, `_snipeit_os_info_6`, `_snipeit_osversion_12`, `_snipeit_mac_address_1`, `_snipeit_bluetooth_mac_address_11` + +These must match your Snipe-IT custom field IDs or assets will be created with missing data. + +## Development Commands + +### Setup +```bash +# Install dependencies +pip3 install -r requirements.txt + +# Copy and configure settings +cp settings_example.ini settings.ini +# Edit settings.ini with your Mosyle and Snipe-IT credentials and configuration +``` + +### Running + +#### One-Time Sync +```bash +# Full sync of configured device types (runs once and exits) +python3 main.py + +# With custom config and logging directory +python3 main.py --config /path/to/settings.ini --log-dir ./logs --log-level INFO + +# Enable debug logging +python3 main.py --log-level DEBUG + +# Backfill Apple model images +python3 appleInfo.py +``` + +#### Daemon Mode (Continuous Loop) +```bash +# Run continuously with 1-hour interval (3600 seconds) +python3 main.py --daemon --interval 3600 + +# Run with 30-minute interval +python3 main.py --daemon --interval 1800 + +# Exit with Ctrl+C +``` + +#### Command-Line Arguments +- `--daemon`: Run continuously in loop mode instead of one-time execution +- `--interval SECONDS`: Time between runs in daemon mode (default: 3600 = 1 hour) +- `--config FILE`: Path to settings.ini (default: settings.ini) +- `--log-dir DIR`: Directory for log files (default: logs) +- `--log-level LEVEL`: Logging verbosity (DEBUG, INFO, WARNING, ERROR, CRITICAL; default: INFO) + +### Scheduled Deployment (systemd) + +The recommended production approach uses Linux systemd with a timer for reliable scheduled execution. + +#### Installation + +```bash +# 1. Prepare settings and credentials +cp settings_example.ini ~/mosyle-config/settings.ini +# Edit ~/mosyle-config/settings.ini with your Mosyle and Snipe-IT credentials + +# 2. Run installation script as root +sudo bash install_systemd.sh ~/mosyle-config/settings.ini +``` + +This script will: +- Create a `mosyle-snipe` system user +- Create `/opt/mosyle-snipe-sync` application directory +- Create `/etc/mosyle-snipe-sync` config directory +- Create `/var/log/mosyle-snipe-sync` log directory +- Set up Python virtual environment with dependencies +- Install systemd service and timer files +- Set proper file permissions + +#### Managing the Service + +```bash +# Enable timer to run at boot +sudo systemctl enable mosyle-snipe-sync.timer + +# Start the timer +sudo systemctl start mosyle-snipe-sync.timer + +# Check timer status +sudo systemctl status mosyle-snipe-sync.timer +systemctl list-timers mosyle-snipe-sync.timer + +# View upcoming run times +sudo systemctl list-timers --all mosyle-snipe-sync.timer + +# Stop the timer +sudo systemctl stop mosyle-snipe-sync.timer + +# Run immediately (don't wait for next scheduled time) +sudo systemctl start mosyle-snipe-sync.service +``` + +#### Modifying the Schedule + +The default schedule is **every 1 hour**, with the first run 10 minutes after boot. + +To change the interval, edit the timer file: + +```bash +sudo nano /etc/systemd/system/mosyle-snipe-sync.timer +``` + +Common OnUnitActiveSec values: +- `1h`: Every 1 hour +- `6h`: Every 6 hours +- `12h`: Every 12 hours +- `1d`: Every 24 hours + +To run at a specific time (e.g., 2 AM daily): + +```bash +# Replace "OnUnitActiveSec=1h" with: +OnCalendar=02:00 +``` + +After editing, reload and restart: + +```bash +sudo systemctl daemon-reload +sudo systemctl restart mosyle-snipe-sync.timer +``` + +#### Viewing Logs + +```bash +# View systemd journal (real-time) +sudo journalctl -u mosyle-snipe-sync.service -f + +# View last 100 lines +sudo journalctl -u mosyle-snipe-sync.service -n 100 + +# View logs from last hour +sudo journalctl -u mosyle-snipe-sync.service --since "1 hour ago" + +# View log files directly +tail -f /var/log/mosyle-snipe-sync/mosyle_snipe_sync.log + +# View with grep filter +journalctl -u mosyle-snipe-sync.service | grep "ERROR" +``` + +### Testing Considerations +- Mosyle API is destructive (can remove ADMIN rights) — test carefully on non-prod first +- Rate limiting is enforced by Snipe-IT (default 120 req/min) — script auto-sleeps when limit is hit +- Serial number must exist in Mosyle for device to sync (devices without serial are skipped) +- User assignment requires matching email between Mosyle and Snipe-IT user accounts +- When testing with systemd, check both `journalctl` and `/var/log/mosyle-snipe-sync/` for full logs + +## Common Development Tasks + +### Adding Support for a New Device Type +1. Ensure Mosyle supports the device type (`mac`, `ios`, `tvos` are supported) +2. Add new category ID to `settings.ini` +3. Add corresponding `createXxxModel()` method in `snipe.py` +4. Update device type loop in `main.py` to call appropriate model creation + +### Mapping Additional Mosyle Fields to Snipe-IT +1. Identify Snipe-IT custom field ID for the target field +2. Add mapping in `snipe.py:buildPayloadFromMosyle()` with hardcoded field ID +3. Update `settings.ini` `[api-mapping]` section for documentation + +### Debugging API Issues +- Enable debug output: Check colorama colored print statements in code +- Mosyle auth fails: Verify admin credentials in `settings.ini` [mosyle] section and that admin has API access +- Snipe-IT rate limit hit: Script automatically waits 60s; check if genuine API usage is exceeding limit +- Model/asset creation fails: Verify category IDs and manufacturer ID exist in Snipe-IT + +## Known Issues & Limitations + +1. **Hardcoded Custom Field IDs**: The `_snipeit_*` field IDs in `snipe.py:buildPayloadFromMosyle()` must match your Snipe-IT instance exactly. No configuration-driven approach currently exists. + +2. **BYOB Logic**: User-enrolled devices in Mosyle are assumed to be BYOB and skipped entirely. May not fit all organizations. + +3. **Timestamp Calltype**: The "timestamp" mode in `settings.ini` for incremental syncs is noted as potentially non-functional. + +4. **Print Verbosity**: Original codebase has extensive print statements for debugging. Production users may find output verbose. + +5. **No Transaction Rollback**: If sync fails mid-process, partial updates remain. No atomic transaction handling. + +## Security Considerations + +- Store Mosyle and Snipe-IT credentials in `settings.ini`, never commit credentials to version control +- Ensure `settings.ini` has restrictive file permissions (e.g., `chmod 600 settings.ini`) +- Snipe-IT API key should be restricted to read/write asset and model operations +- Rate limiting is built-in but ensure your Snipe-IT instance is properly configured +- AppleDB image fetching is over HTTPS but integrates external image data + +## Deployment & Operations + +### Logging System + +The refactored codebase includes comprehensive logging with file rotation: + +- **Logs directory**: `logs/` by default (configurable with `--log-dir`) +- **Log file**: `mosyle_snipe_sync.log` +- **Rotation**: Files rotate at 10MB, keeping 10 backups +- **Format**: `[YYYY-MM-DD HH:MM:SS] LEVEL message` +- **Levels**: DEBUG, INFO, WARNING, ERROR, CRITICAL + +When deployed with systemd: +- Logs go to both file (`/var/log/mosyle-snipe-sync/`) and systemd journal +- Use `journalctl` for real-time monitoring +- Log directory permissions are set to 755 + +### Failure Handling + +- **Non-fatal errors**: Device processing errors are logged and skipped; sync continues with remaining devices +- **API connection failures**: Logged and retried according to Snipe-IT retry logic (429/5xx errors) +- **Missing config/credentials**: Script exits with fatal error and logs details +- **Daemon mode**: Errors in one run don't stop the daemon; next run occurs at scheduled interval + +### Operational Monitoring + +**Key metrics in logs:** +- Total devices processed per run +- Devices created vs. updated +- User assignment changes +- Asset tag synchronization events +- API errors and rate limit hits + +**Health checks:** +```bash +# Is the timer enabled and running? +sudo systemctl is-enabled mosyle-snipe-sync.timer +sudo systemctl is-active mosyle-snipe-sync.timer + +# When will the next run occur? +sudo systemctl list-timers mosyle-snipe-sync.timer + +# Did the last run succeed? +sudo journalctl -u mosyle-snipe-sync.service -n 50 | grep "Synchronization run complete" +``` + +## Dependencies + +- **colorama**: Terminal color output +- **requests**: HTTP library for API calls +- **rich**: Progress bars and console formatting +- **configparser**: Built-in, parse `settings.ini` +- **logging**: Built-in, structured logging with rotation diff --git a/appleInfo.py b/appleInfo.py index 20bddc7..23a0ed7 100644 --- a/appleInfo.py +++ b/appleInfo.py @@ -1,77 +1,80 @@ -#this file can be run to update your Snipe-IT models without interacting with Mosyle. - -import base64 -from tracemalloc import stop import requests import json import datetime import configparser -import colorama -from sys import exit - -from mosyle import Mosyle +from colorama import Fore, Style, init from snipe import Snipe -from colorama import Fore -from colorama import Style -from operator import mod -modelNumber = "iPad11,2"; +# Initialize colorama for colored terminal output +init() -# Converts datetim/e to timestamp for Mosyle -ts = datetime.datetime.now().timestamp() - 200 - -# Set some Variables from the settings.conf: +# Load config config = configparser.ConfigParser() config.read('settings.ini') -# This is the address, cname, or FQDN for your snipe-it instance. snipe_url = config['snipe-it']['url'] apiKey = config['snipe-it']['apiKey'] defaultStatus = config['snipe-it']['defaultStatus'] -apple_manufacturer_id = config['snipe-it']['manufacturer_id'] +apple_manufacturer_id = int(config['snipe-it']['manufacturer_id']) macos_category_id = config['snipe-it']['macos_category_id'] -ios_category_id = config['snipe-it']['ios_category_id'] -tvos_category_id = config['snipe-it']['tvos_category_id'] +ios_category_id = config['snipe-it']['ios_category_id'] +tvos_category_id = config['snipe-it']['tvos_category_id'] macos_fieldset_id = config['snipe-it']['macos_fieldset_id'] ios_fieldset_id = config['snipe-it']['ios_fieldset_id'] tvos_fieldset_id = config['snipe-it']['tvos_fieldset_id'] -deviceTypes = config['mosyle']['deviceTypes'].split(',') - snipe_rate_limit = int(config['snipe-it']['rate_limit']) - apple_image_check = config['snipe-it'].getboolean('apple_image_check') -#setup the snipe-it api -snipe = Snipe(apiKey,snipe_url,apple_manufacturer_id,macos_category_id,ios_category_id,tvos_category_id,snipe_rate_limit, macos_fieldset_id, ios_fieldset_id, tvos_fieldset_id,apple_image_check) +# Initialize Snipe API +snipe = Snipe(apiKey, snipe_url, apple_manufacturer_id, macos_category_id, ios_category_id, tvos_category_id, + snipe_rate_limit, macos_fieldset_id, ios_fieldset_id, tvos_fieldset_id, apple_image_check) + +# Fetch all models +try: + response = snipe.listAllModels() + models = response.json() +except Exception as e: + print(Fore.RED + f"Failed to get models: {e}" + Style.RESET_ALL) + exit(1) +if 'rows' not in models: + print(Fore.RED + "No models found in response." + Style.RESET_ALL) + exit(1) -#get all models -models = snipe.listAllModels().json() -print(models); -#loop through each model +# Process models for model in models['rows']: - #is the model's manufacturer Apple? - print('Processing model: ' + str(model['id']), model["model_number"]) - print("Is the model's manufacturer Apple?", "checking manufacture id " + str(model['manufacturer']['id']) +" against known apple manufacturer id: "+ str(apple_manufacturer_id)) - if int(model['manufacturer']['id']) == int(apple_manufacturer_id): - #yes! - print(Fore.GREEN, "Yes! Checking for photo!", Style.RESET_ALL); - #Does it need a picture? - if model['image'] == None: - print("No photo. Dowloading photos") - imageResponse = snipe.getImageForModel(model["model_number"]); - if imageResponse != False: - print("Photo Downloaded") - snipe.setImageForModel(model["id"],imageResponse.content) - payload = { - "image": imageResponse - } - - snipe.updateModel(str(model['id']), payload) + model_id = model.get('id') + model_name = model.get("model_number") or model.get("name", "Unknown") + print(f"Processing model: {model_id} {model_name}") + + manufacturer = model.get('manufacturer') + if not manufacturer or 'id' not in manufacturer: + print(Fore.YELLOW + f"Model {model_id} has no manufacturer info. Skipping." + Style.RESET_ALL) + continue + + manufacturer_id = int(manufacturer['id']) + print(f"Is the model's manufacturer Apple? checking manufacturer id {manufacturer_id} against {apple_manufacturer_id}") + + if manufacturer_id != apple_manufacturer_id: + print(Fore.YELLOW + "Model is not Apple. Skipping." + Style.RESET_ALL) + continue + + print(Fore.GREEN + "Yes! Checking for photo..." + Style.RESET_ALL) + + if not model.get('image'): + print("No photo found. Attempting download...") + + try: + image_response = snipe.getImageForModel(model_name) + if image_response: + print(Fore.CYAN + "Photo downloaded. Updating model..." + Style.RESET_ALL) + + payload = {"image": image_response} + snipe.updateModel(str(model_id), payload) else: - print("no photo found, moving on") - else: - print("picture already set. Skipping") - else: - print(Fore.YELLOW,'model is not apple. Skip.',Style.RESET_ALL) + print(Fore.YELLOW + f"No photo found for model {model_name}. Skipping." + Style.RESET_ALL) + except Exception as e: + print(Fore.RED + f"Error downloading image for model {model_name}: {e}" + Style.RESET_ALL) + else: + print("Picture already set. Skipping.") diff --git a/install_systemd.sh b/install_systemd.sh new file mode 100644 index 0000000..3737fb1 --- /dev/null +++ b/install_systemd.sh @@ -0,0 +1,146 @@ +#!/bin/bash +# Installation script for MosyleSnipeSync systemd deployment +# Run as root: sudo bash install_systemd.sh [/path/to/config] + +set -e + +# Colors for output +RED='\033[0;31m' +GREEN='\033[0;32m' +YELLOW='\033[1;33m' +NC='\033[0m' # No Color + +# Default paths +APP_DIR="/opt/mosyle-snipe-sync" +CONFIG_DIR="/etc/mosyle-snipe-sync" +LOG_DIR="/var/log/mosyle-snipe-sync" +USER="mosyle-snipe" +GROUP="mosyle-snipe" + +# Check if running as root +if [[ $EUID -ne 0 ]]; then + echo -e "${RED}This script must be run as root (use sudo)${NC}" + exit 1 +fi + +echo -e "${GREEN}=== MosyleSnipeSync systemd Installation ===${NC}" + +# Step 1: Check if settings.ini is provided +if [ $# -lt 1 ]; then + echo -e "${YELLOW}Usage: sudo bash install_systemd.sh /path/to/settings.ini${NC}" + echo "Example: sudo bash install_systemd.sh ~/mosyle-config/settings.ini" + exit 1 +fi + +SETTINGS_FILE="$1" + +if [ ! -f "$SETTINGS_FILE" ]; then + echo -e "${RED}Error: settings.ini not found at $SETTINGS_FILE${NC}" + exit 1 +fi + +# Step 2: Create system user and group +echo -e "${GREEN}Creating system user and group...${NC}" +if id "$USER" &>/dev/null; then + echo "User $USER already exists" +else + useradd --system --shell /bin/false --home-dir "$APP_DIR" "$USER" || echo "User $USER already exists" +fi + +# Step 3: Create necessary directories +echo -e "${GREEN}Creating directories...${NC}" +mkdir -p "$APP_DIR" +mkdir -p "$CONFIG_DIR" +mkdir -p "$LOG_DIR" + +# Step 4: Copy application files +echo -e "${GREEN}Copying application files...${NC}" +SCRIPT_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )" +cp "$SCRIPT_DIR"/*.py "$APP_DIR/" || true +cp "$SCRIPT_DIR"/requirements.txt "$APP_DIR/" 2>/dev/null || true + +# Step 5: Copy and configure settings.ini +echo -e "${GREEN}Copying configuration file...${NC}" +cp "$SETTINGS_FILE" "$CONFIG_DIR/settings.ini" +chmod 600 "$CONFIG_DIR/settings.ini" + +# Step 6: Copy .env file if it exists +if [ -f "$SCRIPT_DIR/.env" ]; then + echo -e "${GREEN}Copying .env file...${NC}" + cp "$SCRIPT_DIR/.env" "$APP_DIR/.env" + chmod 600 "$APP_DIR/.env" +else + echo -e "${YELLOW}Warning: .env file not found. You'll need to create it manually.${NC}" + echo "Create $APP_DIR/.env with your Mosyle credentials:" + cat > "$APP_DIR/.env.example" << EOF +url=https://businessapi.mosyle.com/v1 +token=your_token_here +user=admin_email@example.com +password=admin_password +EOF + chmod 600 "$APP_DIR/.env.example" + echo "Example .env file created at $APP_DIR/.env.example" +fi + +# Step 7: Set permissions +echo -e "${GREEN}Setting permissions...${NC}" +chown -R "$USER:$GROUP" "$APP_DIR" +chown -R "$USER:$GROUP" "$CONFIG_DIR" +chown -R "$USER:$GROUP" "$LOG_DIR" +chmod 755 "$APP_DIR" +chmod 755 "$CONFIG_DIR" +chmod 755 "$LOG_DIR" + +# Step 8: Create Python virtual environment +echo -e "${GREEN}Creating Python virtual environment...${NC}" +python3 -m venv "$APP_DIR/venv" +"$APP_DIR/venv/bin/pip" install --upgrade pip setuptools wheel > /dev/null 2>&1 + +if [ -f "$APP_DIR/requirements.txt" ]; then + echo "Installing Python dependencies..." + "$APP_DIR/venv/bin/pip" install -r "$APP_DIR/requirements.txt" +else + echo -e "${YELLOW}requirements.txt not found, installing manually${NC}" + "$APP_DIR/venv/bin/pip" install colorama requests rich python-dotenv +fi + +# Step 9: Install systemd service and timer +echo -e "${GREEN}Installing systemd service and timer...${NC}" +cp "$SCRIPT_DIR/systemd/mosyle-snipe-sync.service" /etc/systemd/system/ +cp "$SCRIPT_DIR/systemd/mosyle-snipe-sync.timer" /etc/systemd/system/ + +# Reload systemd daemon +systemctl daemon-reload + +# Step 10: Final instructions +echo "" +echo -e "${GREEN}=== Installation Complete ===${NC}" +echo "" +echo "Next steps:" +echo "1. Verify configuration:" +echo " cat $CONFIG_DIR/settings.ini" +echo "" +echo "2. If you haven't done so, create the .env file with Mosyle credentials:" +echo " sudo cp $APP_DIR/.env.example $APP_DIR/.env" +echo " sudo nano $APP_DIR/.env" +echo " sudo chown $USER:$GROUP $APP_DIR/.env" +echo " sudo chmod 600 $APP_DIR/.env" +echo "" +echo "3. Enable and start the timer:" +echo " sudo systemctl enable mosyle-snipe-sync.timer" +echo " sudo systemctl start mosyle-snipe-sync.timer" +echo "" +echo "4. Monitor the service:" +echo " sudo systemctl status mosyle-snipe-sync.timer" +echo " sudo systemctl list-timers mosyle-snipe-sync.timer" +echo "" +echo "5. View logs:" +echo " sudo journalctl -u mosyle-snipe-sync.service -f" +echo " or" +echo " tail -f $LOG_DIR/mosyle_snipe_sync.log" +echo "" +echo "6. To change the run interval, edit the timer:" +echo " sudo nano /etc/systemd/system/mosyle-snipe-sync.timer" +echo " sudo systemctl daemon-reload" +echo " sudo systemctl restart mosyle-snipe-sync.timer" +echo "" diff --git a/logger_config.py b/logger_config.py new file mode 100644 index 0000000..cde44aa --- /dev/null +++ b/logger_config.py @@ -0,0 +1,60 @@ +""" +Logging configuration for MosyleSnipeSync. +Sets up structured logging with file and console handlers. +""" +import logging +import logging.handlers +import os +from pathlib import Path + + +def setup_logging(log_dir="logs", log_level="INFO"): + """ + Configure logging with file rotation and console output. + + Args: + log_dir: Directory to store log files (created if doesn't exist) + log_level: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL) + """ + # Create logs directory if it doesn't exist + log_path = Path(log_dir) + log_path.mkdir(exist_ok=True) + + # Create logger + logger = logging.getLogger("mosyle_snipe_sync") + logger.setLevel(getattr(logging, log_level)) + + # Remove any existing handlers to avoid duplicates + logger.handlers.clear() + + # File handler with rotation (10MB, keep 10 files) + file_handler = logging.handlers.RotatingFileHandler( + log_path / "mosyle_snipe_sync.log", + maxBytes=10 * 1024 * 1024, # 10MB + backupCount=10 + ) + file_handler.setLevel(getattr(logging, log_level)) + + # Console handler for stderr + console_handler = logging.StreamHandler() + console_handler.setLevel(getattr(logging, log_level)) + + # Formatter with timestamp (using {}-style to safely handle % characters in messages) + formatter = logging.Formatter( + "[{asctime}] {levelname:<8} {message}", + datefmt="%Y-%m-%d %H:%M:%S", + style="{" + ) + file_handler.setFormatter(formatter) + console_handler.setFormatter(formatter) + + # Add handlers to logger + logger.addHandler(file_handler) + logger.addHandler(console_handler) + + return logger + + +def get_logger(): + """Get the configured logger instance.""" + return logging.getLogger("mosyle_snipe_sync") diff --git a/main.py b/main.py index 1faa5a7..b9288af 100644 --- a/main.py +++ b/main.py @@ -1,167 +1,340 @@ -# Import all the things +""" +MosyleSnipeSync main script. +Synchronizes Apple device data from Mosyle to Snipe-IT. +Can run as a one-time sync or as a scheduled daemon. +""" import json import datetime import configparser -import colorama -from sys import exit +import argparse +import time +import sys +import os +from pathlib import Path +from rich.progress import Progress +from rich.console import Console from mosyle import Mosyle from snipe import Snipe -from colorama import Fore -from colorama import Style +from logger_config import setup_logging, get_logger -# Converts datetim/e to timestamp for Mosyle -ts = datetime.datetime.now().timestamp() - 200 -# Set some Variables from the settings.conf: -config = configparser.ConfigParser() -config.read('settings.ini') +def load_configuration(config_file='settings.ini'): + """Load configuration from settings.ini.""" + logger = get_logger() -# This is the address, cname, or FQDN for your snipe-it instance. -snipe_url = config['snipe-it']['url'] -apiKey = config['snipe-it']['apiKey'] -defaultStatus = config['snipe-it']['defaultStatus'] -apple_manufacturer_id = config['snipe-it']['manufacturer_id'] -macos_category_id = config['snipe-it']['macos_category_id'] -ios_category_id = config['snipe-it']['ios_category_id'] -tvos_category_id = config['snipe-it']['tvos_category_id'] -macos_fieldset_id = config['snipe-it']['macos_fieldset_id'] -ios_fieldset_id = config['snipe-it']['ios_fieldset_id'] -tvos_fieldset_id = config['snipe-it']['tvos_fieldset_id'] -deviceTypes = config['mosyle']['deviceTypes'].split(',') + # Load configuration file + if not Path(config_file).exists(): + logger.error(f"Configuration file not found: {config_file}") + raise FileNotFoundError(f"Configuration file not found: {config_file}") -snipe_rate_limit = int(config['snipe-it']['rate_limit']) + config = configparser.ConfigParser(interpolation=None) + config.read(config_file) -apple_image_check = config['snipe-it'].getboolean('apple_image_check') + # Extract Mosyle config + try: + mosyle_url = config['mosyle']['url'] + mosyle_token = config['mosyle']['token'] + mosyle_user = config['mosyle']['user'] + mosyle_password = config['mosyle']['password'] + deviceTypes = config['mosyle']['deviceTypes'].split(',') + calltype = config['mosyle'].get('calltype', 'all') + except KeyError as e: + logger.error(f"Missing required Mosyle configuration: {e}") + raise ValueError(f"Missing required Mosyle configuration: {e}") + # Verify required Mosyle credentials + if not all([mosyle_url, mosyle_token, mosyle_user, mosyle_password]): + logger.error("Missing required Mosyle credentials in settings.ini") + raise ValueError("Missing Mosyle credentials in settings.ini [mosyle] section") + # Extract Snipe-IT config + try: + snipe_url = config['snipe-it']['url'] + apiKey = config['snipe-it']['apiKey'] + apple_manufacturer_id = config['snipe-it']['manufacturer_id'] + macos_category_id = config['snipe-it']['macos_category_id'] + ios_category_id = config['snipe-it']['ios_category_id'] + tvos_category_id = config['snipe-it']['tvos_category_id'] + macos_fieldset_id = config['snipe-it']['macos_fieldset_id'] + ios_fieldset_id = config['snipe-it']['ios_fieldset_id'] + tvos_fieldset_id = config['snipe-it']['tvos_fieldset_id'] + snipe_rate_limit = int(config['snipe-it']['rate_limit']) + apple_image_check = config['snipe-it'].getboolean('apple_image_check') + except KeyError as e: + logger.error(f"Missing required configuration key: {e}") + raise -# Set the token for the Mosyle Api -mosyle = Mosyle(config['mosyle']['token'], config['mosyle']['url'], config['mosyle']['user'], config['mosyle']['password']) + logger.info("Configuration loaded successfully") -# Set the call type for Mosyle -calltype = config['mosyle']['calltype'] + return { + 'mosyle': { + 'url': mosyle_url, + 'token': mosyle_token, + 'user': mosyle_user, + 'password': mosyle_password, + 'deviceTypes': deviceTypes, + 'calltype': calltype + }, + 'snipe': { + 'url': snipe_url, + 'apiKey': apiKey, + 'manufacturer_id': apple_manufacturer_id, + 'macos_category_id': macos_category_id, + 'ios_category_id': ios_category_id, + 'tvos_category_id': tvos_category_id, + 'macos_fieldset_id': macos_fieldset_id, + 'ios_fieldset_id': ios_fieldset_id, + 'tvos_fieldset_id': tvos_fieldset_id, + 'rate_limit': snipe_rate_limit, + 'apple_image_check': apple_image_check + } + } -#setup the snipe-it api -snipe = Snipe(apiKey,snipe_url,apple_manufacturer_id,macos_category_id,ios_category_id,tvos_category_id,snipe_rate_limit, macos_fieldset_id, ios_fieldset_id, tvos_fieldset_id,apple_image_check) -for deviceType in deviceTypes: - # Get the list of devices from Mosyle based on the deviceType and call type +def run_sync(config): + """ + Execute a single synchronization run. - if calltype == "timestamp": - mosyle_response = mosyle.listTimestamp(ts, ts, deviceType).json() - else: - mosyle_response = mosyle.list(deviceType).json() - - #print(mosyle_response) - if 'status' in mosyle_response: - if mosyle_response['status'] != "OK": - print('There was an issue with the Mosyle API. Stopping.', mosyle_response['message']) - exit(); - if 'status' in mosyle_response['response'][0]: - print('There was an issue with the Mosyle API. Stopping script.') - print(mosyle_response['response'][0]['info']) - exit() + Args: + config: Configuration dictionary from load_configuration() - + Returns: + int: Total number of devices processed + """ + logger = get_logger() + console = Console() + logger.info("=== Starting synchronization run ===") + try: + # Initialize Mosyle + mosyle = Mosyle( + config['mosyle']['token'], + config['mosyle']['user'], + config['mosyle']['password'], + config['mosyle']['url'] + ) + logger.info("Successfully connected to Mosyle") + except Exception as e: + logger.error(f"Failed to connect to Mosyle: {e}") + raise - print('starting snipe') + try: + # Initialize Snipe-IT + snipe = Snipe( + config['snipe']['apiKey'], + config['snipe']['url'], + config['snipe']['manufacturer_id'], + config['snipe']['macos_category_id'], + config['snipe']['ios_category_id'], + config['snipe']['tvos_category_id'], + config['snipe']['rate_limit'], + config['snipe']['macos_fieldset_id'], + config['snipe']['ios_fieldset_id'], + config['snipe']['tvos_fieldset_id'], + config['snipe']['apple_image_check'] + ) + logger.info("Successfully connected to Snipe-IT") + except Exception as e: + logger.error(f"Failed to connect to Snipe-IT: {e}") + raise + total_devices_processed = 0 + ts = datetime.datetime.now().timestamp() - 200 - print('Looping through Mosyle Hardware List') - # Return Mosyle hardware and search them in snipe - for sn in mosyle_response['response'][0]['devices']: - print('Sarting for Mosyle Device ', sn['device_name']) - if sn['serial_number'] == None: - print('There is no serial number here. It must be user enrolled?') - #print(sn) + for deviceType in config['mosyle']['deviceTypes']: + deviceType = deviceType.strip() + logger.info(f"Processing device type: {deviceType}") + + try: + # Fetch devices from Mosyle + if config['mosyle']['calltype'] == "timestamp": + logger.debug(f"Using timestamp mode for {deviceType}") + mosyle_response = mosyle.listTimestamp(ts, ts, deviceType) + else: + logger.debug(f"Using 'all' mode for {deviceType} (paginated)") + all_devices = [] + page = 1 + while True: + response = mosyle.list(deviceType, page=page) + devices = response.get('response', {}).get('devices', []) + if not devices: + break + all_devices.extend(devices) + logger.debug(f"Retrieved {len(devices)} devices from page {page}") + page += 1 + mosyle_response = {"status": "OK", "response": {"devices": all_devices}} + + if mosyle_response.get('status') != "OK": + logger.error(f"Mosyle API error for {deviceType}: {mosyle_response.get('message')}") + continue + + devices = mosyle_response['response'].get('devices', []) + device_count = len(devices) + logger.info(f"Found {device_count} {deviceType} devices in Mosyle") + + # Process each device + with Progress() as progress: + task = progress.add_task(f"[green]Processing {deviceType} devices...", total=device_count) + + for device_index, sn in enumerate(devices, 1): + try: + if sn['serial_number'] is None: + logger.warning(f"{deviceType} device at index {device_index} has no serial number, skipping") + progress.advance(task) + continue + + # Look up existing asset + asset = snipe.listHardware(sn['serial_number']).json() + + # Look up or create model + model = snipe.searchModel(sn['device_model']).json() + if model['total'] == 0: + logger.info(f"Creating new model: {sn['device_model']}") + if sn['os'] == "mac": + model = snipe.createModel(sn['device_model']).json()['payload']['id'] + elif sn['os'] == "ios": + model = snipe.createMobileModel(sn['device_model']).json()['payload']['id'] + elif sn['os'] == "tvos": + model = snipe.createAppleTvModel(sn['device_model']).json()['payload']['id'] + else: + model = model['rows'][0]['id'] + + # Check for assigned user + mosyle_user = sn.get('useremail') if sn.get('CurrentConsoleManagedUser') and 'useremail' in sn else None + devicePayload = snipe.buildPayloadFromMosyle(sn) + + # Create asset if doesn't exist + if asset.get('total', 0) == 0: + logger.info(f"Creating new asset: {sn['serial_number']} ({sn['device_model']})") + asset = snipe.createAsset(model, devicePayload) + if mosyle_user: + logger.info(f"Assigning asset to user: {mosyle_user}") + snipe.assignAsset(mosyle_user, asset['payload']['id']) + total_devices_processed += 1 + progress.advance(task) + continue + + # Update existing asset + if asset.get('total') == 1 and asset.get('rows'): + logger.info(f"Updating asset: {sn['serial_number']}") + snipe.updateAsset(asset['rows'][0]['id'], devicePayload, model) + + # Sync user assignment + if mosyle_user: + assigned = asset['rows'][0]['assigned_to'] + if assigned is None and sn.get('useremail'): + logger.info(f"Assigning asset to user: {sn['useremail']}") + snipe.assignAsset(sn['useremail'], asset['rows'][0]['id']) + elif sn.get('useremail') is None: + logger.info(f"Unassigning asset: {asset['rows'][0]['id']}") + snipe.unasigneAsset(asset['rows'][0]['id']) + elif assigned and assigned['username'] != sn['useremail']: + logger.info(f"Reassigning asset from {assigned['username']} to {sn['useremail']}") + snipe.unasigneAsset(asset['rows'][0]['id']) + snipe.assignAsset(sn['useremail'], asset['rows'][0]['id']) + + # Sync asset tag back to Mosyle + asset_tag = asset['rows'][0].get('asset_tag') if asset.get('rows') else None + if not sn.get('asset_tag') or sn['asset_tag'] != asset_tag: + if asset_tag: + logger.info(f"Syncing asset tag to Mosyle: {sn['serial_number']} -> {asset_tag}") + mosyle.setAssetTag(sn['serial_number'], asset_tag) + + total_devices_processed += 1 + progress.advance(task) + + except Exception as e: + logger.error(f"Error processing device {sn.get('serial_number', 'unknown')}: {e}") + progress.advance(task) + continue + + logger.info(f"Finished {deviceType}: {total_devices_processed} total devices processed") + + except Exception as e: + logger.error(f"Error processing device type {deviceType}: {e}") continue - else: - print('Device has serial number! ',str(sn['serial_number'])) - - print('Checking snipe for Mosyle device by serial number: '+str(sn['serial_number'])) - asset = snipe.listHardware(sn['serial_number']).json() - - #check to see if Device model already exists on snipe - - print("Checking to see if device model already exist on SnipeIt:", sn['device_model']) - model = snipe.searchModel(sn['device_model']).json() - print("Model:", model) - # Create the asset model if is not exist - if model['total'] == 0: - print('Model does not exist in Snipe. Need to make it.') - if sn['os'] == "mac": - print('Making a new Mac model', sn['device_model']) - model = snipe.createModel(sn['device_model']).json() - model = model['payload']['id'] - if sn['os'] == "ios": - print('Making a new ios model', sn['device_model']) - model = snipe.createMobileModel(sn['device_model']).json() - model = model['payload']['id'] - if sn['os'] == "tvos": - print('Making New Apple TV Model', sn['device_model']) - model = snipe.createAppleTvModel(sn['device_model']).json() - model = model['payload']['id'] - else: - print('Model already exists in SnipeIt!') - model = model['rows'][0]['id'] + logger.info(f"=== Synchronization run complete. Total devices processed: {total_devices_processed} ===") + return total_devices_processed - - if sn['CurrentConsoleManagedUser'] != None and "userid" in sn: - mosyle_user = sn['userid'] - else: - print('this device is not currently assigned. Dont try to assign it later'); - mosyle_user = None - - - #Create payload translating Mosyle to SnipeIt - devicePayload = snipe.buildPayloadFromMosyle(sn); - - # If asset doesnt exist create and assign it - if asset['total'] == 0: - asset = snipe.createAsset(model, devicePayload).json() - if mosyle_user != None: - print('Assigning asset to SnipIT user based on Mosyle Assignment') - snipe.assignAsset(mosyle_user, asset['payload']['id']) - continue +def main(): + """Main entry point supporting both one-time and daemon modes.""" + parser = argparse.ArgumentParser( + description='Synchronize Apple devices from Mosyle to Snipe-IT' + ) + parser.add_argument( + '--daemon', + action='store_true', + help='Run in daemon mode (continuously loop)' + ) + parser.add_argument( + '--interval', + type=int, + default=3600, + help='Interval between runs in seconds (default: 3600 = 1 hour). Only used in daemon mode.' + ) + parser.add_argument( + '--config', + default='settings.ini', + help='Path to settings.ini file (default: settings.ini)' + ) + parser.add_argument( + '--log-level', + default='INFO', + choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'], + help='Logging level (default: INFO)' + ) + parser.add_argument( + '--log-dir', + default='logs', + help='Directory for log files (default: logs)' + ) - # Update existing Devices - print(asset) - if asset['total'] == 1: - #f"{x:.2f}" - print('Asset ', sn['serial_number'],' already exists in SnipeIt. Update it.') - print(asset['rows'][0]['name']) - snipe.updateAsset(asset['rows'][0]['id'], devicePayload) - - # Check the asset assignement state - if mosyle_user != None: - if asset['rows'][0]['assigned_to'] == None and sn['userid'] != None: - snipe.assignAsset(sn['userid'], asset['rows'][0]['id']) - #continue - - elif sn['userid'] == None: - snipe.unasigneAsset(asset['rows'][0]['id']) - #continue - - elif asset['rows'][0]['assigned_to']['username'] == sn['userid']: - print('nothing to see here') - elif asset['rows'][0]['assigned_to']['username'] != sn['userid']: - snipe.unasigneAsset(asset['rows'][0]['id']) - snipe.assignAsset(sn['userid'], asset['rows'][0]['id']) - else: - print('no assignement actions') - - print("Checking to see if Mosyle needs an updated asset tag") - #if there is no asset tag on mosyle, add the snipeit asset tag - if(sn['asset_tag'] == None or sn['asset_tag'] == "" or sn['asset_tag'] != asset['rows'][0]['asset_tag']): - print('update the mosyle asset tag of device ', sn['serial_number'], 'to ', asset['rows'][0]['asset_tag']) - mosyle.setAssetTag(sn['serial_number'], asset['rows'][0]['asset_tag']) + args = parser.parse_args() + + # Setup logging + setup_logging(log_dir=args.log_dir, log_level=args.log_level) + logger = get_logger() + + logger.info("MosyleSnipeSync started") + logger.info(f"Mode: {'daemon' if args.daemon else 'one-time'}") + if args.daemon: + logger.info(f"Interval: {args.interval} seconds ({args.interval / 3600:.1f} hours)") + + try: + # Load configuration + config = load_configuration(args.config) + + if args.daemon: + # Daemon mode: run continuously + logger.info("Entering daemon mode") + run_count = 0 + while True: + try: + run_count += 1 + logger.info(f"--- Run {run_count} ---") + run_sync(config) + logger.info(f"Sleeping for {args.interval} seconds") + time.sleep(args.interval) + except KeyboardInterrupt: + logger.info("Received interrupt signal, exiting daemon mode") + break + except Exception as e: + logger.error(f"Error in daemon run {run_count}: {e}") + logger.info(f"Sleeping for {args.interval} seconds before retry") + time.sleep(args.interval) else: - print('Mosyle already has an assest tag of: ', sn['asset_tag']) - - print('Finished with OS: ', deviceType) - print('') \ No newline at end of file + # One-time mode: run once and exit + run_sync(config) + logger.info("Exiting") + + except Exception as e: + logger.error(f"Fatal error: {e}") + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/mosyle.py b/mosyle.py index cee9bb8..c04b1be 100644 --- a/mosyle.py +++ b/mosyle.py @@ -1,64 +1,65 @@ -import base64 import requests class Mosyle: - - # Create Mosyle instance - def __init__(self, key, url = "https://businessapi.mosyle.com/v1", user = "", password = ""): - # Attribute the variable to the instance - self.url = url - self.request = requests.Session() - self.request.headers["accesstoken"] = key - #base64 encode username and password for basic auth - userpass = user + ':' + password - encoded_u = base64.b64encode(userpass.encode()).decode() - self.request.headers["Authorization"] = "Basic %s" % encoded_u + def __init__(self, access_token, email, password, url="https://managerapi.mosyle.com/v2"): + self.url = url + self.access_token = access_token + self.email = email + self.password = password + self.session = requests.Session() + self.jwt_token = self.login() - - # Create variables requests - def list(self, os): - print("Listing devices for OS:", os) - params = { - "operation": "list", - "options": { - "os": os - } - } - # Concatanate url and send the request - return self.request.post(self.url + "/devices", json = params ) + if self.jwt_token: + self.session.headers.update({ + "Authorization": f"Bearer {self.jwt_token}", + "Content-Type": "application/json" + }) + else: + raise Exception("Login failed. Could not obtain JWT token.") - def listTimestamp(self, start, end, os): - params = { - "operation": "list", - "options": { - "os": os, - "enrolldate_start": start, - "enrolldate_end": end - } - } - return self.request.post(self.url + "/devices", json = params ) + def login(self): + payload = { + "accessToken": self.access_token, + "email": self.email, + "password": self.password + } + response = self.session.post(f"{self.url}/login", json=payload) + + if response.status_code == 200: + auth_header = response.headers.get("Authorization", "") + if auth_header.startswith("Bearer "): + return auth_header.replace("Bearer ", "") + else: + print("Authorization header missing or malformed.") + else: + print(f"Login failed. Status Code: {response.status_code}") + print(f"Error: {response.text}") + return None - def listmobile(self): - params = { + def _post(self, endpoint, data): + data["accessToken"] = self.access_token + response = self.session.post(f"{self.url}/{endpoint}", json=data) + try: + return response.json() + except Exception: + return {"error": "Invalid JSON response", "text": response.text} + + def list(self, os, specific_columns=None, page=1): + print("Listing devices for OS:", os, "Page:", page) + data = { + "accessToken": self.access_token, "operation": "list", "options": { - "os": "ios" + "os": os, + "page": page } } - return self.request.post(self.url + "/devices", json = params ) - - def listuser(self, iduser): - params = { - "operation": "list_users", - "options": { "identifiers": [iduser] - } - } - return self.request.post(self.url + "/users", json = params ) - - def setAssetTag(self, serialnumber, tag): - params = { + if specific_columns: + data["specific_columns"] = specific_columns + return self._post("listdevices", data) + def setAssetTag(self, serialnumber, tag): + return self._post("devices", { "operation": "update_device", "serialnumber": serialnumber, "asset_tag": tag - } - return self.request.post(self.url + "/devices", json = params ) \ No newline at end of file + }) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..e667473 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +colorama==0.4.6 +requests==2.31.0 +rich==13.7.0 diff --git a/settings_example.ini b/settings_example.ini index afe35ad..1b7db34 100644 --- a/settings_example.ini +++ b/settings_example.ini @@ -44,3 +44,13 @@ apple_image_check = True #leftside is the snipe-it field name, rightside is the mosyle field name name = general name _snipeit_mac_address_1 = general mac_address + +[logging] +#Directory where log files will be stored (created if doesn't exist) +log_dir = logs +#Logging level: DEBUG, INFO, WARNING, ERROR, CRITICAL (default: INFO) +log_level = INFO +#Maximum size of each log file in MB before rotation (default: 10) +log_max_size_mb = 10 +#Number of old log files to keep (default: 10) +log_backup_count = 10 diff --git a/snipe.py b/snipe.py index 63366fa..3257500 100644 --- a/snipe.py +++ b/snipe.py @@ -1,4 +1,3 @@ -from cgi import print_arguments import mimetypes from unittest import result import requests @@ -38,40 +37,37 @@ def listHardware(self, serial): def listAllModels(self): print('requesting all apple models') - return self.snipeItRequest("GET","/models", params = {"limit": "50", "offset": "0", "sort": "created_at", "order": "asc"}) + return self.snipeItRequest("GET","/models", params = {"limit": "200", "offset": "0", "sort": "created_at", "order": "asc"}) def searchModel(self, model): print('Requesting Snipe Model list') - result = self.snipeItRequest("GET", "/models", params = {"limit": "50", "offset": "0", "search": model, "sort": "created_at", "order": "asc"}) - print(result.json()) + result = self.snipeItRequest("GET", "/models", params={ + "limit": "50", "offset": "0", "search": model, "sort": "created_at", "order": "asc" + }) jsonResult = result.json() - #Did the search return a result? + if jsonResult['total'] == 0: - print("model was not found") + print("Model was not found.") else: - print("the model was found") - - #does the model have a picture? - if jsonResult['rows'][0]['image'] is None: - print("the model does not have a picture. Let, set one") - #No, it does not. Let's update it. - imageResponse = self.getImageForModel(model); - print("imageResponse", imageResponse) - if(imageResponse == False): - print("loading the image failed..") + print("Model was found.") + model_data = jsonResult['rows'][0] + + if model_data['image'] is None: + print("The model does not have a picture. Let's set one.") + image_data_url = self.getImageForModel(model) + + if not image_data_url: + print("Failed to get image.") else: payload = { - "image": imageResponse + "image": image_data_url } - self.updateModel(str(jsonResult['rows'][0]['id']), payload) - - + self.updateModel(str(model_data['id']), payload) else: - print('image already set.'); - - #print(result) + print("Image already set.") + return result - + def createModel(self, model): imageResponse = self.getImageForModel(model); @@ -83,7 +79,7 @@ def createModel(self, model): "category_id": self.macos_category_id, "manufacturer_id": self.manufacturer_id, "model_number": model, - "fieldset_id": self.mac_os_fieldset_id, + "fieldset_id": self.macos_fieldset_id, "image":imageResponse } @@ -97,41 +93,52 @@ def createAsset(self, model, payload): print(payload); payload['status_id'] = 2 payload['model_id'] = model + payload['asset_tag'] = payload['serial'] - asset = self.snipeItRequest("POST", "/hardware", json = payload).json() #print(asset) - payload = { - "serial": payload['serial'] - } - return self.snipeItRequest("PATCH", "/hardware/" + str(asset['payload']['id']), json = payload) - + return self.snipeItRequest("POST", "/hardware", json = payload).json() def assignAsset(self, user, asset_id): print('Assigning asset '+str(asset_id)+' to user '+user) - + email_to_match = user.lower() + payload = { - "search": user, - "limit": 2 + "search": email_to_match, + "limit": 10 # Increase in case multiple matches exist } - response = self.snipeItRequest("GET", "/users", params = payload).json() - + response = self.snipeItRequest("GET", "/users", params=payload).json() + print(f"{response} Payload: {payload}") if response['total'] == 0: return + # Find exact email match + user_row = next((row for row in response['rows'] if row['email'].lower() == email_to_match), None) + + if not user_row: + print(f"No exact match found for {email_to_match}") + return + payload = { - "assigned_user": response['rows'][0]['id'], + "assigned_user": user_row['id'], "checkout_to_type": "user" } - return self.snipeItRequest("POST", "/hardware/" + str(asset_id) + "/checkout", json = payload) + return self.snipeItRequest("POST", f"/hardware/{asset_id}/checkout", json=payload) + def unasigneAsset(self, asset_id): print('Unassigning asset '+str(asset_id)) return self.snipeItRequest("POST", "/hardware/" + str(asset_id) + "/checkin") - def updateAsset(self, asset_id, payload): - print('Updating asset '+str(asset_id)) - #print(payload) - return self.snipeItRequest("PATCH", "/hardware/" + str(asset_id), json = payload) + def updateAsset(self, asset_id, payload, model_id=None): + print('Updating asset ' + str(asset_id)) + payload = dict(payload) # Make a copy to avoid mutating the original + payload.pop('serial', None) + + if model_id: + payload['model_id'] = model_id # Include model assignment + + return self.snipeItRequest("PATCH", "/hardware/" + str(asset_id), json=payload) + def createMobileModel(self, model): print('creating new mobile Model') @@ -171,7 +178,7 @@ def buildPayloadFromMosyle(self, payload): #"asset_tag": asset, "name": payload['device_name'], "serial": payload['serial_number'], - "_snipeit_bluetooth_mac_address_8": payload['bluetooth_mac_address'] + "_snipeit_bluetooth_mac_address_11": payload['bluetooth_mac_address'] } #lets get the proper os name @@ -192,10 +199,10 @@ def buildPayloadFromMosyle(self, payload): os = "Not Known" - finalPayload['_snipeit_operating_system_3'] = os + finalPayload['_snipeit_os_info_6'] = os #set os version - finalPayload['_snipeit_operating_system_version_4'] = payload['osversion'] + finalPayload['_snipeit_osversion_12'] = payload['osversion'] #macaddress stuff wifiMac = payload['wifi_mac_address'] @@ -209,50 +216,121 @@ def buildPayloadFromMosyle(self, payload): return finalPayload - def snipeItRequest(self, type, url, params = None, json = None): - self.request_count += 1 - if(self.request_count >= self.rate_limit): - print(Fore.YELLOW + "Max requests per minute reached. Sleeping for 60 seconds") - time.sleep(60) - self.request_count = 0 - print(Fore.GREEN + "Request count has been reset", "Continuing", Style.RESET_ALL) - - - if(type == "GET"): - print('Sending GET request to snipeit', url) - return requests.get(self.url + url, headers = self.headers, params = params) - elif(type == "POST"): - print('Sending POST request to snipeit', url) - return requests.post(self.url + url, headers = self.headers, json = json) - elif(type == "PATCH"): - print('Sending PATCH request to snipeit', url) - return requests.patch(self.url + url, headers = self.headers, json = json) - elif(type == "DELETE"): - print('Sending DELETE request to snipeit', url) - return requests.delete(self.url + url, headers = self.headers) - else: - print(Fore.RED+'Unknown request type'+Style.RESET_ALL) - return None + def snipeItRequest(self, type, url, params=None, json=None): + max_retries = 5 + retry_delay = 60 # seconds - def getImageForModel(self, modelNumber): - if self.apple_image_check == True: + for attempt in range(max_retries): + if self.request_count >= self.rate_limit: + print(Fore.YELLOW + "Max requests per minute reached. Sleeping for 60 seconds..." + Style.RESET_ALL) + time.sleep(60) + self.request_count = 0 - url = "https://img.appledb.dev/device@512/" + modelNumber + "/0.png" - print("Get image from URL", url) try: - response = requests.get(url) - response.raise_for_status() - base64encoded = base64.b64encode(response.content).decode("utf8") - fullImageSring = "data:image/png;name=0.png;base64,"+ base64encoded; - return fullImageSring; - - - except requests.exceptions.HTTPError as err: - print(Fore.RED + "Error getting image from apple db", err, Style.RESET_ALL) - return False - else: + self.request_count += 1 + print(f'Sending {type} request to Snipe-IT: {url}') + + if type == "GET": + response = requests.get(self.url + url, headers=self.headers, params=params) + elif type == "POST": + response = requests.post(self.url + url, headers=self.headers, json=json) + elif type == "PATCH": + response = requests.patch(self.url + url, headers=self.headers, json=json) + elif type == "DELETE": + response = requests.delete(self.url + url, headers=self.headers) + else: + print(Fore.RED + 'Unknown request type' + Style.RESET_ALL) + return None + + if response.status_code == 429: + print(Fore.YELLOW + f"Rate limited by server (429). Waiting {retry_delay} seconds before retrying..." + Style.RESET_ALL) + time.sleep(retry_delay) + continue + + if response.status_code >= 500: + print(Fore.RED + f"Server error {response.status_code}. Retrying in {retry_delay} seconds..." + Style.RESET_ALL) + time.sleep(retry_delay) + self.request_count = 0 + continue + + return response + + except requests.RequestException as e: + print(Fore.RED + f"Request failed: {e}. Retrying in {retry_delay} seconds..." + Style.RESET_ALL) + time.sleep(retry_delay) + + print(Fore.RED + f"Failed to complete request after {max_retries} attempts: {url}" + Style.RESET_ALL) + return None + + + def getImageForModel(self, model_number): + if not self.apple_image_check: print("Image checking is disabled.") return False + + print(f"Trying to look up model info from AppleDB: {model_number}") + try: + response = requests.get("https://api.appledb.dev/device/main.json") + response.raise_for_status() + devices = response.json() + + for device in devices: + identifiers = device.get("identifier", []) + device_maps = device.get("deviceMap", []) + + if model_number in device_maps or model_number in identifiers: + device_key = device.get("key", model_number) + colors = device.get("colors", []) + color = colors[0]["key"] if colors and isinstance(colors[0], dict) and "key" in colors[0] else "Silver" + + image_url = f"https://img.appledb.dev/device@256/{device_key}/{color}.png" + print(f"Found match. Trying image URL: {image_url}") + + img_response = requests.get(image_url) + img_response.raise_for_status() + + base64encoded = base64.b64encode(img_response.content).decode("utf8") + full_image_string = "data:image/png;name=image.png;base64," + base64encoded + return full_image_string + + print(f"No matching identifier or deviceMap found for {model_number}") + + except requests.exceptions.RequestException as e: + print(Fore.RED + f"Error getting image from AppleDB: {e}" + Style.RESET_ALL) + except Exception as e: + print(Fore.RED + f"Unexpected error during AppleDB lookup: {e}" + Style.RESET_ALL) + + return False + + + + def setImageForModel(self, model_id, image_bytes): + """ + Uploads an image to a model in Snipe-IT. + + :param model_id: ID of the model in Snipe-IT + :param image_bytes: Raw image bytes (from requests.get().content) + """ + url = f"{self.url}/models/{model_id}" + headers = { + "Authorization": f"Bearer {self._snipetoken}" + } + files = { + "image": ("image.png", image_bytes, "image/png") + } + + try: + response = requests.post(url, headers=headers, files=files) + response.raise_for_status() + print(Fore.GREEN + f"Successfully uploaded image for model ID {model_id}" + Style.RESET_ALL) + return response + except requests.RequestException as e: + print(Fore.RED + f"Failed to upload image to model {model_id}: {e}" + Style.RESET_ALL) + return None + + + + #if __name__ == "__main__": diff --git a/systemd/mosyle-snipe-sync.service b/systemd/mosyle-snipe-sync.service new file mode 100644 index 0000000..bee83a5 --- /dev/null +++ b/systemd/mosyle-snipe-sync.service @@ -0,0 +1,20 @@ +[Unit] +Description=MosyleSnipeSync - Synchronize devices from Mosyle to Snipe-IT +Documentation=https://github.com/your-org/MosyleSnipeSync +Wants=network-online.target +After=network-online.target + +[Service] +Type=oneshot +User=mosyle-snipe +Group=mosyle-snipe +WorkingDirectory=/opt/mosyle-snipe-sync +ExecStart=/opt/mosyle-snipe-sync/venv/bin/python3 /opt/mosyle-snipe-sync/main.py --config /etc/mosyle-snipe-sync/settings.ini --log-dir /var/log/mosyle-snipe-sync +Environment="PYTHONUNBUFFERED=1" +StandardOutput=journal +StandardError=journal +SyslogIdentifier=mosyle-snipe-sync + +# Prevent concurrent runs +Type=oneshot +KillMode=process diff --git a/systemd/mosyle-snipe-sync.timer b/systemd/mosyle-snipe-sync.timer new file mode 100644 index 0000000..2e9412d --- /dev/null +++ b/systemd/mosyle-snipe-sync.timer @@ -0,0 +1,22 @@ +[Unit] +Description=MosyleSnipeSync Timer - Run every 1 hour +Documentation=https://github.com/your-org/MosyleSnipeSync +Requires=mosyle-snipe-sync.service + +[Timer] +# Run every 1 hour +OnBootSec=10min +OnUnitActiveSec=1h + +# Optional: Run at a specific time (e.g., 2 AM every day) +# Uncomment the line below and comment out OnUnitActiveSec to use this instead +# OnCalendar=02:00 + +# If a run is missed, run it when the system comes back online +Persistent=true + +# Allow randomization to prevent thundering herd (randomize up to 5 minutes) +RandomizedDelaySec=5min + +[Install] +WantedBy=timers.target