Signed-off-by: Maciej Pijanowski <maciej.pijanowski@3mdeb.com>
This commit is contained in:
Maciej Pijanowski
2023-06-05 22:30:10 +02:00
parent b3346fb121
commit 7dbdc9faaf
7 changed files with 370 additions and 16 deletions
+4 -16
View File
@@ -1,21 +1,9 @@
MIT License
Copyright (c) 2023 Dasharo
Copyright (c) 2023 3mdeb <contact@3mdeb.com>
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+17
View File
@@ -0,0 +1,17 @@
# OSVF scripts
This repository gathers various scripts related to the Dasharo Open Source
Firmware Validation in some ways.
## Scripts
* [snipeit](snipeit/README.md) - manage devices in SnipeIT via API
## Contributing
Contributions are welcome! If you encounter any issues, have suggestions, or
want to contribute improvements, please open an issue or submit a pull request.
## License
This project is licensed under the [MIT License](../LICENSE).
+6
View File
@@ -0,0 +1,6 @@
INSTALL_PATH = /usr/local/bin/snipeit
CONFIG_FILE = $(HOME)/.osfv/snipeit.yml
install:
sudo install -m 755 snipeit.py /usr/local/bin/snipeit
test -f $(HOME)/.osfv/snipeit.yml || install -D -m 644 config.yml $(HOME)/.osfv/snipeit.yml
+101
View File
@@ -0,0 +1,101 @@
# Snipe-IT Asset Retrieval
This script allows you to interact with Snipe-IT, an open-source asset
management system, via its API. You can use this script to retrieve information
about assets, list used and unused assets, check out and check in assets, and
more.
This script is specific to the OSFV environment and Snipe-IT configuration, and
as such, may not fit for other needs.
## Installation
1. Clone the repository to your local machine:
```shell
git clone https://github.com/dasharo/osfv-scripts.git
```
1. Navigate to the cloned repository:
```shell
cd osfv-scripts/snipeit
```
1. Install the necessary dependencies. Make sure you have Python 3 and pip
installed, then run:
```shell
pip install -r requirements.txt
```
1. Install the script and config file template:
```shell
make install
```
4. Customize the configuration:
- Open `~/.osfv/snipeit.yml` and provide your Snipe-IT API URL, API token,
and user ID
## Usage
To use the script, you can run it with different commands and options. Here are
some examples:
- List all used assets:
```shell
snipeit.py list_used
```
- List all unused assets:
```shell
snipeit.py list_unused
```
- List all assets:
```shell
snipeit.py list_all
```
- Check out an asset (by asset ID):
```shell
snipeit.py check_out --asset_id 123
```
- Check in an asset (by asset ID):
```shell
snipeit.py check_in --asset_id 123
```
- Check out an asset (by RTE IP):
```bash
snipeit check_out --rte_ip <rte_ip_address>
```
- Check out an asset (by RTE IP):
```bash
snipeit check_out --rte_ip <rte_ip_address>
```
> Replace `<rte_ip_address>` with the actual RTE IP address of the asset you
> want to check out. The script will identify the asset based on the RTE IP and
> perform the check-out process.
> Please note that the RTE IP should match the value stored in the asset's
> custom field named "RTE IP".
- For more command options, you can use the `--help` flag:
```shell
snipeit.py --help
```
+3
View File
@@ -0,0 +1,3 @@
api_url: 'http://snipeit/api/v1'
api_token: 'YOUR_PERSONAL_API_TOKEN'
user_id: YOUR_USER_ID
+3
View File
@@ -0,0 +1,3 @@
PyYAML==6.0
PyYAML==6.0
Requests==2.31.0
+236
View File
@@ -0,0 +1,236 @@
#!/usr/bin/env python3
import argparse
import requests
import yaml
import os
import sys
CONFIG_FILE_PATH = os.path.expanduser('~/.osfv/snipeit.yml')
# Retrieve API configuration from YAML file
def load_api_config():
try:
with open(CONFIG_FILE_PATH, 'r') as file:
config = yaml.safe_load(file)
except FileNotFoundError:
raise FileNotFoundError(f'Configuration file not found')
except yaml.YAMLError as e:
raise ValueError(f'Error parsing YAML: {e}')
if config is None:
raise ValueError(f'Empty configuration file')
sys.exit(1)
api_url = config.get('api_url')
api_token = config.get('api_token')
user_id = config.get('user_id')
if not api_url or not api_token:
raise ValueError('Incomplete API configuration in the YAML file')
return api_url, api_token, user_id
try:
# API endpoint and authentication token
api_url, api_token, user_id = load_api_config()
except FileNotFoundError as e:
print(f'Configuration file not found at {CONFIG_FILE_PATH}: {e}')
sys.exit(1)
except ValueError as e:
print(f'Please check the {CONFIG_FILE_PATH} file: {e}')
sys.exit(1)
# Headers for API requests
headers = {
'Accept': 'application/json',
'Authorization': f'Bearer {api_token}'
}
# Retrieve all assets
def get_all_assets():
page = 1
all_assets = []
while True:
response = requests.get(f'{api_url}/hardware', headers=headers, params={'limit': 500, 'offset': (page - 1) * 500}, timeout=10)
if response.status_code == 200:
data = response.json()
all_assets.extend(data['rows'])
if 'total_pages' not in data or data['total_pages'] <= page:
break
page += 1
else:
print(f'Error retrieving assets. Status code: {response.status_code}')
print(response.json())
break
return all_assets
# List used assets
def list_used_assets():
all_assets = get_all_assets()
used_assets = [asset for asset in all_assets if asset['assigned_to'] is not None]
if used_assets:
for asset in used_assets:
print_asset_details(asset)
else:
print('No used assets found.')
# List unused assets
def list_unused_assets():
all_assets = get_all_assets()
unused_assets = [asset for asset in all_assets if asset['assigned_to'] is None]
if unused_assets:
for asset in unused_assets:
print_asset_details(asset)
else:
print('No unused assets found.')
# List all assets
def list_all_assets():
all_assets = get_all_assets()
if all_assets:
for asset in all_assets:
print_asset_details(asset)
else:
print('No assets found.')
# Print asset details as JSON with specific custom fields
def list_for_zabbix():
all_assets = get_all_assets()
if all_assets:
for asset in all_assets:
print_asset_details_for_zabbix(asset)
else:
print('No assets found.')
# Print asset details including custom fields
def print_asset_details(asset):
print(f'Asset Tag: {asset["asset_tag"]}, Asset ID: {asset["id"]}, Name: {asset["name"]}, Serial: {asset["serial"]}')
custom_fields = asset.get('custom_fields', {})
if custom_fields:
for field_name, field_data in custom_fields.items():
field_value = field_data.get('value')
print(f'{field_name}: {field_value}')
print()
def get_asset_id_by_rte_ip(rte_ip):
# Retrieve all assets
all_assets = get_all_assets()
# Search for asset with matching RTE IP
for asset in all_assets:
custom_fields = asset.get('custom_fields', {})
if custom_fields:
rte_ip_field = next((field_data['value'] for field_name, field_data in custom_fields.items() if field_name == 'RTE IP'), None)
if rte_ip_field == rte_ip:
return asset['id']
# No asset found with matching RTE IP
return None
# Print asset details formatted as an input for Zabbix import script
def print_asset_details_for_zabbix(asset):
output = {}
custom_fields = asset.get('custom_fields', {})
if custom_fields:
for field_name, field_data in custom_fields.items():
if field_name in ["RTE IP", "Sonoff IP", "PiKVM IP"]:
field_value = field_data.get('value')
if field_value:
key = f'{asset["asset_tag"]}_{field_name}'.replace(' ', '_')
output[key] = field_value
print(f'{key}: {output[key]}')
# Check out an asset
def check_out_asset(asset_id):
data = {
'asset_id': asset_id,
'assigned_user': user_id,
'checkout_to_type': 'user'
}
response = requests.post(f'{api_url}/hardware/{asset_id}/checkout', headers=headers, json=data, timeout=10)
if response.status_code == 200:
print(f'Asset {asset_id} successfully checked out to {user_id} user.')
else:
print(f'Error checking out asset {asset_id} to user {user_id}. Status code: {response.status_code}')
print(response.json())
# Check in an asset
def check_in_asset(asset_id):
response = requests.post(f'{api_url}/hardware/{asset_id}/checkin', headers=headers, timeout=10)
if response.status_code == 200:
print(f'Asset {asset_id} successfully checked in.')
else:
print(f'Error checking in asset {asset_id}. Status code: {response.status_code}')
print(response.json())
# Main function
def main():
parser = argparse.ArgumentParser(description='Snipe-IT Asset Retrieval')
subparsers = parser.add_subparsers(title='commands', dest='command', help='Command to execute')
list_used_parser = subparsers.add_parser('list_used', help='List all already used assets')
list_unused_parser = subparsers.add_parser('list_unused', help='List all unused assets')
list_all_parser = subparsers.add_parser('list_all', help='List all assets')
list_zabbix_parser = subparsers.add_parser('list_for_zabbix', help='List assets in a format suitable for Zabbix integration')
check_out_parser = subparsers.add_parser('check_out', help='Check out an asset by providing the Asset ID or RTE IP')
check_out_group = check_out_parser.add_mutually_exclusive_group(required=True)
check_out_group.add_argument('--asset_id', type=int, help='Asset ID')
check_out_group.add_argument('--rte_ip', type=str, help='RTE IP')
check_in_parser = subparsers.add_parser('check_in', help='Check in an asset by providing the Asset ID or RTE IP')
check_in_group = check_in_parser.add_mutually_exclusive_group(required=True)
check_in_group.add_argument('--asset_id', type=int, help='Asset ID')
check_in_group.add_argument('--rte_ip', type=str, help='RTE IP')
args = parser.parse_args()
if args.command == 'list_used':
list_used_assets()
elif args.command == 'list_unused':
list_unused_assets()
elif args.command == 'list_all':
list_all_assets()
elif args.command == 'list_for_zabbix':
list_for_zabbix()
elif args.command == 'check_out':
if args.asset_id:
check_out_asset(args.asset_id)
elif args.rte_ip:
# Use the RTE IP to find the asset ID and perform check out
asset_id = get_asset_id_by_rte_ip(args.rte_ip)
if asset_id:
check_out_asset(asset_id)
else:
print(f'No asset found with RTE IP: {args.rte_ip}')
elif args.command == 'check_in':
if args.asset_id:
check_in_asset(args.asset_id)
elif args.rte_ip:
# Use the RTE IP to find the asset ID and perform check in
asset_id = get_asset_id_by_rte_ip(args.rte_ip)
if asset_id:
check_in_asset(asset_id)
else:
print(f'No asset found with RTE IP: {args.rte_ip}')
if __name__ == '__main__':
main()