CI/CD hardening: lint, secret scan, coverage gate, auto CVE-fix PRs, GHCR + GitHub mirror publishing (#32)
CI / build-and-scan (push) Successful in 1m37s

- release changelog: commits rendered as description (link), divider lines dropped
- gitleaks secret scan and hadolint on every push/PR
- ruff lint/format gate (Python repos) with a pytest --cov-fail-under gate
- scheduled CRITICAL Trivy failures attempt an apk upgrade rebuild and open a follow-up PR if it clears the finding, instead of just failing red
- images also published to ghcr.io/djeex/<repo>
- a matching GitHub Release is created on the GitHub mirror, with a notice pointing back to this repo as the source of truth
This commit was merged in pull request #32.
This commit is contained in:
2026-08-26 15:43:40 +02:00
parent b3ed21eec2
commit 5a7a60d299
10 changed files with 356 additions and 135 deletions
+42 -31
View File
@@ -1,11 +1,13 @@
import json
import logging
import os
import re
import logging
import json
import sys
# Read version from VERSION file
with open(os.path.join(os.path.dirname(os.path.dirname(__file__)), "VERSION"), "r", encoding="utf-8") as f:
with open(
os.path.join(os.path.dirname(os.path.dirname(__file__)), "VERSION"), encoding="utf-8"
) as f:
VERSION = f.read().strip()
# Logger setup
@@ -23,13 +25,13 @@ logging.info("=" * 60)
# Env variables
try:
DISCORD_WEBHOOK_URL = os.environ['DISCORD_WEBHOOK_URL']
DISCORD_SERVER_NAME = os.environ.get('DISCORD_SERVER_NAME', 'Shared for free')
DISCORD_ROLES = os.environ.get('DISCORD_ROLES')
COUNTRY = os.environ.get('COUNTRY') or 'US'
REFRESH_TIME = int(os.environ.get('REFRESH_TIME') or 30)
TEST_MODE = os.environ.get('TEST_MODE', 'False').lower() == 'true'
PRODUCT_NAMES = os.environ['PRODUCT_NAMES']
DISCORD_WEBHOOK_URL = os.environ["DISCORD_WEBHOOK_URL"]
DISCORD_SERVER_NAME = os.environ.get("DISCORD_SERVER_NAME", "Shared for free")
DISCORD_ROLES = os.environ.get("DISCORD_ROLES")
COUNTRY = os.environ.get("COUNTRY") or "US"
REFRESH_TIME = int(os.environ.get("REFRESH_TIME") or 30)
TEST_MODE = os.environ.get("TEST_MODE", "False").lower() == "true"
PRODUCT_NAMES = os.environ["PRODUCT_NAMES"]
# Errors and warning
except KeyError as e:
@@ -49,32 +51,32 @@ if not DISCORD_WEBHOOK_URL:
logging.error("❌ DISCORD_WEBHOOK_URL is required but not defined.")
sys.exit(1)
PRODUCT_NAMES = [name.strip() for name in PRODUCT_NAMES.split(',')]
PRODUCT_NAMES = [name.strip() for name in PRODUCT_NAMES.split(",")]
# Role mapping
DISCORD_ROLE_MAP = {}
if not DISCORD_ROLES or not DISCORD_ROLES.strip():
logging.warning("⚠️ DISCORD_ROLES not defined or empty. Defaulting all roles to @everyone.")
for name in PRODUCT_NAMES:
DISCORD_ROLE_MAP[name] = '@everyone'
DISCORD_ROLE_MAP[name] = "@everyone"
else:
roles = [r.strip() if r.strip() else '@everyone' for r in DISCORD_ROLES.split(',')]
roles = [r.strip() if r.strip() else "@everyone" for r in DISCORD_ROLES.split(",")]
if len(roles) != len(PRODUCT_NAMES):
logging.error("❌ The number of DISCORD_ROLES must match PRODUCT_NAMES.")
sys.exit(1)
for name, role in zip(PRODUCT_NAMES, roles):
if role != '@everyone' and not re.match(r'^<@&\d{17,20}>$', role):
if role != "@everyone" and not re.match(r"^<@&\d{17,20}>$", role):
logging.error(f"❌ Invalid DISCORD_ROLE format for {name}: {role}")
sys.exit(1)
DISCORD_ROLE_MAP[name] = role
# Masked webhook in terminal
match = re.search(r'/(\d+)/(.*)', DISCORD_WEBHOOK_URL)
match = re.search(r"/(\d+)/(.*)", DISCORD_WEBHOOK_URL)
if match:
webhook_id = match.group(1)
webhook_token = match.group(2)
masked_webhook_id = webhook_id[:len(webhook_id) - 10] + '*' * 10
masked_webhook_token = webhook_token[:len(webhook_token) - 120] + '*' * 10
masked_webhook_id = webhook_id[: len(webhook_id) - 10] + "*" * 10
masked_webhook_token = webhook_token[: len(webhook_token) - 120] + "*" * 10
wh_masked_url = f"https://discord.com/api/webhooks/{masked_webhook_id}/{masked_webhook_token}"
else:
wh_masked_url = "[Invalid webhook URL]"
@@ -90,29 +92,33 @@ HEADERS = {
"Connection": "keep-alive",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Ch-Ua": "\"Google Chrome\";v=\"131\", \"Chromium\";v=\"131\", \"Not.A/Brand\";v=\"24\"",
"Sec-Ch-Ua-Platform": "\"macOS\"",
"Sec-Ch-Ua": '"Google Chrome";v="131", "Chromium";v="131", "Not.A/Brand";v="24"',
"Sec-Ch-Ua-Platform": '"macOS"',
"Cache-Control": "no-cache, no-store, must-revalidate",
"Pragma": "no-cache",
"Expires": "0"
"Expires": "0",
}
# Load country setting and localization config
country_code = os.environ.get("COUNTRY", "US").upper()
try:
with open("localization.json", "r", encoding="utf-8") as f:
with open("localization.json", encoding="utf-8") as f:
localization_config = json.load(f)
except FileNotFoundError:
logging.error("❌ localization.json file not found.")
sys.exit(1)
# Find country entry
country_entry = next((entry for entry in localization_config if entry["country_code"].upper() == country_code), None)
country_entry = next(
(entry for entry in localization_config if entry["country_code"].upper() == country_code), None
)
if not country_entry:
logging.warning(f"⚠️ Country '{country_code}' not found in localization.json. Defaulting to US.")
country_entry = next((entry for entry in localization_config if entry["country_code"].upper() == "US"), None)
country_entry = next(
(entry for entry in localization_config if entry["country_code"].upper() == "US"), None
)
if not country_entry:
logging.error("❌ US fallback not found in localization.json.")
sys.exit(1)
@@ -124,7 +130,7 @@ currency = country_entry["currency"]
# Load language file
try:
with open("languages.json", "r", encoding="utf-8") as f:
with open("languages.json", encoding="utf-8") as f:
loc_lang = json.load(f)
except FileNotFoundError:
logging.error("❌ languages.json file not found.")
@@ -141,9 +147,15 @@ if not loc:
# Ensure all required keys are present
required_keys = [
"in_stock_title", "out_of_stock_title", "sku_change_title",
"buy_now", "price", "time", "footer",
"sku_description", "imminent_drop"
"in_stock_title",
"out_of_stock_title",
"sku_change_title",
"buy_now",
"price",
"time",
"footer",
"sku_description",
"imminent_drop",
]
missing_keys = [key for key in required_keys if key not in loc]
fallback = loc_lang.get("en", {})
@@ -159,17 +171,16 @@ for key in missing_keys:
locale = full_lang_code.lower()
API_URL_SKU = os.getenv(
"API_URL_SKU",
f"https://api.nvidia.partners/edge/product/search?page=1&limit=100&locale={locale}&Manufacturer=Nvidia"
f"https://api.nvidia.partners/edge/product/search?page=1&limit=100&locale={locale}&Manufacturer=Nvidia",
)
API_URL_STOCK = os.getenv(
"API_URL_STOCK",
f"https://api.store.nvidia.com/partner/v1/feinventory?locale={locale}&skus="
"API_URL_STOCK", f"https://api.store.nvidia.com/partner/v1/feinventory?locale={locale}&skus="
)
PRODUCT_URL = os.getenv(
"PRODUCT_URL",
f"https://marketplace.nvidia.com/{locale}/consumer/graphics-cards/?locale={locale}&page=1&limit=12&manufacturer=NVIDIA"
f"https://marketplace.nvidia.com/{locale}/consumer/graphics-cards/?locale={locale}&page=1&limit=12&manufacturer=NVIDIA",
)
# Public constants
+21 -14
View File
@@ -1,21 +1,28 @@
import requests
import logging
import time
from env_config import HEADERS, PRODUCT_NAMES, API_URL_SKU, API_URL_STOCK, PRODUCT_URL
from notifier import send_discord_notification, send_out_of_stock_notification, send_sku_change_notification
import requests
from requests.adapters import HTTPAdapter, Retry
from env_config import API_URL_SKU, API_URL_STOCK, HEADERS, PRODUCT_NAMES, PRODUCT_URL
from notifier import (
send_discord_notification,
send_out_of_stock_notification,
send_sku_change_notification,
)
# HTTP session
session = requests.Session()
retries = Retry(total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504])
session.mount('https://', HTTPAdapter(max_retries=retries))
session.mount("https://", HTTPAdapter(max_retries=retries))
session.headers.update(HEADERS)
# Keeping memory of last run
# Keeping memory of last run
last_sku_dict = {}
global_stock_status_dict = {}
first_run_dict = {name: True for name in PRODUCT_NAMES}
# Stock check function
def check_rtx_50_founders():
global last_sku_dict, global_stock_status_dict, first_run_dict
@@ -24,7 +31,7 @@ def check_rtx_50_founders():
try:
cache_buster = int(time.time() * 1000)
sku_url = f"{API_URL_SKU}&_t={cache_buster}"
response = session.get(sku_url, timeout=10)
logging.info(f"SKU API response: {response.status_code}")
response.raise_for_status()
@@ -32,9 +39,9 @@ def check_rtx_50_founders():
except requests.exceptions.RequestException as e:
logging.error(f"SKU API error: {e}")
return
# Checking productSKU and productUPC for all GPU set in PRODUCT_NAME
all_products = data['searchedProducts']['productDetails']
all_products = data["searchedProducts"]["productDetails"]
for product_name in PRODUCT_NAMES:
product_details = None
@@ -47,8 +54,8 @@ def check_rtx_50_founders():
logging.warning(f"⚠️ No product with GPU '{product_name}' found.")
continue
product_sku = product_details['productSKU']
product_upc = product_details.get('productUPC', "")
product_sku = product_details["productSKU"]
product_upc = product_details.get("productUPC", "")
if not isinstance(product_upc, list):
product_upc = [product_upc]
@@ -60,7 +67,7 @@ def check_rtx_50_founders():
last_sku_dict[product_name] = product_sku
first_run_dict[product_name] = False
# Check product availability in API_URL_STOCK for each SKU
cache_buster = int(time.time() * 1000)
api_stock_url = f"{API_URL_STOCK}{product_sku}&_t={cache_buster}"
@@ -80,8 +87,8 @@ def check_rtx_50_founders():
products_price = "Price not available"
if isinstance(products, list) and len(products) > 0:
for p in products:
price = p.get("price", 'Price not available')
if price != 'Price not available':
price = p.get("price", "Price not available")
if price != "Price not available":
products_price = price
break
else:
@@ -93,7 +100,7 @@ def check_rtx_50_founders():
is_active = p.get("is_active") == "true"
if is_active and any(upc.upper() in gpu_name for upc in product_upc):
found_in_stock.add(gpu_name)
# Comparing previous state and notify
for upc in product_upc:
upc_upper = upc.upper()
+6 -4
View File
@@ -1,17 +1,20 @@
import time
import logging
import signal
import sys
from gpu_checker import check_rtx_50_founders
import time
from env_config import REFRESH_TIME
from gpu_checker import check_rtx_50_founders
# Signal handler function
def handle_exit(signum, frame):
logging.info(f"🛑 Received signal {signum}. Exiting gracefully...")
sys.exit(0)
# Register signal handlers
signal.signal(signal.SIGINT, handle_exit) # Ctrl+C
signal.signal(signal.SIGINT, handle_exit) # Ctrl+C
signal.signal(signal.SIGTERM, handle_exit) # docker stop / kill -15
if __name__ == "__main__":
@@ -24,4 +27,3 @@ if __name__ == "__main__":
except KeyboardInterrupt:
logging.info("🛑 Script interrupted by user (KeyboardInterrupt). Exiting gracefully.")
sys.exit(0)
+43 -19
View File
@@ -1,15 +1,29 @@
import time
import logging
import time
import requests
from env_config import (
DISCORD_WEBHOOK_URL, DISCORD_SERVER_NAME, DISCORD_ROLE_MAP, TEST_MODE, currency,
in_stock_title, out_of_stock_title, sku_change_title,
buy_now, price_label, time_label, footer, sku_description, imminent_drop
DISCORD_ROLE_MAP,
DISCORD_SERVER_NAME,
DISCORD_WEBHOOK_URL,
TEST_MODE,
buy_now,
currency,
footer,
imminent_drop,
in_stock_title,
out_of_stock_title,
price_label,
sku_change_title,
sku_description,
time_label,
)
AVATAR = "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/ds_wh_pp.jpg"
THUMBNAIL = "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/RTX5000.jpg"
# In stock
def send_discord_notification(gpu_name, product_link, products_price):
timestamp = int(time.time())
@@ -24,17 +38,20 @@ def send_discord_notification(gpu_name, product_link, products_price):
"author": {"name": "Nvidia Founder Editions"},
"fields": [
{"name": price_label, "value": f"`{currency}{products_price}`", "inline": True},
{"name": time_label, "value": f"<t:{timestamp}:d> <t:{timestamp}:T>", "inline": True}
{"name": time_label, "value": f"<t:{timestamp}:d> <t:{timestamp}:T>", "inline": True},
],
"description": buy_now.format(product_link=product_link),
"footer": {"text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME), "icon_url": AVATAR}
"footer": {
"text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME),
"icon_url": AVATAR,
},
}
payload = {
"content": DISCORD_ROLE_MAP.get(gpu_name, "@everyone"),
"username": "NviBot",
"avatar_url": AVATAR,
"embeds": [embed]
"embeds": [embed],
}
try:
@@ -46,6 +63,7 @@ def send_discord_notification(gpu_name, product_link, products_price):
except Exception as e:
logging.error(f"🚨 Error sending webhook: {e}")
# Out of stock
def send_out_of_stock_notification(gpu_name, product_link, products_price):
timestamp = int(time.time())
@@ -59,15 +77,16 @@ def send_out_of_stock_notification(gpu_name, product_link, products_price):
"thumbnail": {"url": THUMBNAIL},
"url": product_link,
"author": {"name": "Nvidia Founder Editions"},
"footer": {"text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME), "icon_url": AVATAR},
"fields": [{"name": time_label, "value": f"<t:{timestamp}:d> <t:{timestamp}:T>", "inline": True}]
"footer": {
"text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME),
"icon_url": AVATAR,
},
"fields": [
{"name": time_label, "value": f"<t:{timestamp}:d> <t:{timestamp}:T>", "inline": True}
],
}
payload = {
"username": "NviBot",
"avatar_url": AVATAR,
"embeds": [embed]
}
payload = {"username": "NviBot", "avatar_url": AVATAR, "embeds": [embed]}
try:
response = requests.post(DISCORD_WEBHOOK_URL, json=payload)
@@ -78,6 +97,7 @@ def send_out_of_stock_notification(gpu_name, product_link, products_price):
except Exception as e:
logging.error(f"🚨 Error sending webhook: {e}")
# SKU change
def send_sku_change_notification(gpu_name, old_sku, new_sku, product_link):
timestamp = int(time.time())
@@ -90,15 +110,20 @@ def send_sku_change_notification(gpu_name, old_sku, new_sku, product_link):
"url": product_link,
"description": sku_description.format(old_sku=old_sku, new_sku=new_sku),
"color": 16776960,
"footer": {"text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME), "icon_url": AVATAR},
"fields": [{"name": time_label, "value": f"<t:{timestamp}:d> <t:{timestamp}:T>", "inline": True}]
"footer": {
"text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME),
"icon_url": AVATAR,
},
"fields": [
{"name": time_label, "value": f"<t:{timestamp}:d> <t:{timestamp}:T>", "inline": True}
],
}
payload = {
"content": imminent_drop.format(DISCORD_ROLE=DISCORD_ROLE_MAP.get(gpu_name, '@everyone')),
"content": imminent_drop.format(DISCORD_ROLE=DISCORD_ROLE_MAP.get(gpu_name, "@everyone")),
"username": "NviBot",
"avatar_url": AVATAR,
"embeds": [embed]
"embeds": [embed],
}
try:
@@ -109,4 +134,3 @@ def send_sku_change_notification(gpu_name, old_sku, new_sku, product_link):
logging.error(f"❌ Webhook error: {response.status_code} - {response.text}")
except Exception as e:
logging.error(f"🚨 Error sending webhook: {e}")