Add secret scanning, Dockerfile lint, ruff lint/format gate, coverage gate, and automatic CVE remediation PRs
- gitleaks (via docker cp, dockerignore-agnostic) and hadolint scan every push/PR - new ruff lint stage (ruff.toml pins known-first-party for host/container consistency; B905 in env_config.py's zip() left un-fixed — app-logic change, see feedback-no-app-logic-changes) - pytest --cov-fail-under=75 gate on the test stage - scheduled Trivy critical failures now attempt an apk upgrade rebuild and open a PR if it clears the finding, instead of just failing red - ruff --fix/--format applied to existing code to start the gate clean
This commit is contained in:
+42
-31
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user