austnv/remnawave-minishop
1
0
Fork 0
Code Issues 1 Pull Requests Activity

added happ crypt4 links

Browse Source
This commit is contained in:
machka pasla
2025-12-11 22:17:43 +03:00
parent d0e34126e1
commit a1d3db0e16
17 changed files with 231 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+4
View File
@@ -150,6 +150,10 @@ TRIAL_DURATION_DAYS=5 #
TRIAL_TRAFFIC_LIMIT_GB=0 # Traffic limit for the trial period (0 = unlimited)
TRIAL_TRAFFIC_STRATEGY="NO_RESET" # Traffic reset strategy for the trial period (NO_RESET, WEEK, MONTH)
# Connection link handling (happ crypt4)
CRYPT4_ENABLED=False # Enable happ crypt4 encryption for subscription URLs
CRYPT4_REDIRECT_URL= # Base redirect to wrap the connect button, e.g. https://redir.example.com?url=
# Web Server Settings (for handling webhooks)
WEB_SERVER_HOST="0.0.0.0"
WEB_SERVER_PORT=8080
bot/handlers/user/payment.py
+21 -11
View File
@@ -23,6 +23,7 @@ from config.settings import Settings
from bot.services.notification_service import NotificationService
from bot.keyboards.inline.user_keyboards import get_connect_and_main_keyboard
from bot.utils.text_sanitizer import sanitize_display_name, username_for_display
from bot.utils.config_link import prepare_config_links
payment_processing_lock = asyncio.Lock()
@@ -252,6 +253,10 @@ async def process_successful_payment(session: AsyncSession, bot: Bot,
traffic_label = (
str(int(traffic_amount_gb)) if float(traffic_amount_gb).is_integer() else f"{traffic_amount_gb:g}"
)
config_link_display, connect_button_url = prepare_config_links(
settings, activation_details.get("subscription_url") if activation_details else None
)
config_link_text = config_link_display or _("config_link_not_available")
# For auto-renew charges, avoid re-sending config link; send concise message
if sale_mode != "traffic" and is_auto_renew and final_end_date_for_user:
details_message = _(
@@ -261,21 +266,21 @@ async def process_successful_payment(session: AsyncSession, bot: Bot,
)
details_markup = None
elif sale_mode == "traffic":
config_link = activation_details.get("subscription_url") or _("config_link_not_available")
details_message = _(
"payment_successful_traffic_full",
traffic_gb=traffic_label,
end_date=final_end_date_for_user.strftime('%Y-%m-%d') if final_end_date_for_user else "—",
config_link=config_link,
config_link=config_link_text,
)
details_markup = get_connect_and_main_keyboard(
user_lang, i18n, settings, config_link, preserve_message=True
user_lang,
i18n,
settings,
config_link_display,
connect_button_url=connect_button_url,
preserve_message=True,
)
else:
config_link = activation_details.get("subscription_url") or _(
"config_link_not_available"
)
if applied_referee_bonus_days_from_referral and final_end_date_for_user:
inviter_name_display = _("friend_placeholder")
if db_user and db_user.referred_by_id:
@@ -295,7 +300,7 @@ async def process_successful_payment(session: AsyncSession, bot: Bot,
bonus_days=applied_referee_bonus_days_from_referral,
final_end_date=final_end_date_for_user.strftime('%Y-%m-%d'),
inviter_name=inviter_name_display,
config_link=config_link,
config_link=config_link_text,
)
elif applied_promo_bonus_days > 0 and final_end_date_for_user:
details_message = _(
@@ -303,14 +308,14 @@ async def process_successful_payment(session: AsyncSession, bot: Bot,
months=int(subscription_months),
bonus_days=applied_promo_bonus_days,
end_date=final_end_date_for_user.strftime('%Y-%m-%d'),
config_link=config_link,
config_link=config_link_text,
)
elif final_end_date_for_user:
details_message = _(
"payment_successful_full",
months=int(subscription_months),
end_date=final_end_date_for_user.strftime('%Y-%m-%d'),
config_link=config_link,
config_link=config_link_text,
)
else:
logging.error(
@@ -319,7 +324,12 @@ async def process_successful_payment(session: AsyncSession, bot: Bot,
details_message = _("payment_successful_error_details")
details_markup = get_connect_and_main_keyboard(
user_lang, i18n, settings, config_link, preserve_message=True
user_lang,
i18n,
settings,
config_link_display,
connect_button_url=connect_button_url,
preserve_message=True,
)
try:
await bot.send_message(
bot/handlers/user/promo_user.py
+9 -4
View File
@@ -134,16 +134,21 @@ async def process_promo_code_input(message: types.Message, state: FSMContext,
new_end_date = result if isinstance(result, datetime) else None
active = await subscription_service.get_active_subscription_details(session, user.id)
config_link = active.get("config_link") if active else None
config_link = config_link or _("config_link_not_available")
config_link_display = active.get("config_link") if active else None
connect_button_url = active.get("connect_button_url") if active else None
config_link_text = config_link_display or _("config_link_not_available")
response_to_user_text = _(
"promo_code_applied_success_full",
end_date=(new_end_date.strftime("%d.%m.%Y %H:%M:%S") if new_end_date else "N/A"),
config_link=config_link,
config_link=config_link_text,
)
reply_markup = get_connect_and_main_keyboard(
current_lang, i18n, settings, config_link
current_lang,
i18n,
settings,
config_link_display,
connect_button_url=connect_button_url,
)
else:
await session.rollback()
bot/handlers/user/start.py
+11 -4
View File
@@ -478,21 +478,28 @@ async def start_command_handler(message: types.Message,
# Get updated subscription details
active = await subscription_service.get_active_subscription_details(session, user_id)
config_link = active.get("config_link") if active else None
config_link = config_link or _("config_link_not_available")
config_link_display = active.get("config_link") if active else None
connect_button_url = active.get("connect_button_url") if active else None
config_link_text = config_link_display or _("config_link_not_available")
new_end_date = result if isinstance(result, datetime) else None
promo_success_text = _(
"promo_code_applied_success_full",
end_date=(new_end_date.strftime("%d.%m.%Y %H:%M:%S") if new_end_date else "N/A"),
config_link=config_link,
config_link=config_link_text,
)
from bot.keyboards.inline.user_keyboards import get_connect_and_main_keyboard
await message.answer(
promo_success_text,
reply_markup=get_connect_and_main_keyboard(current_lang, i18n, settings, config_link),
reply_markup=get_connect_and_main_keyboard(
current_lang,
i18n,
settings,
config_link_display,
connect_button_url=connect_button_url,
),
parse_mode="HTML"
)
bot/handlers/user/subscription/core.py
+6 -3
View File
@@ -142,6 +142,9 @@ async def my_subscription_command_handler(
end_date = active.get("end_date")
days_left = (end_date.date() - datetime.now().date()).days if end_date else 0
traffic_mode = bool(getattr(settings, "traffic_sale_mode", False))
config_link_display = active.get("config_link")
connect_button_url = active.get("connect_button_url")
config_link_value = config_link_display or get_text("config_link_not_available")
def _fmt_gb(val: Optional[float]) -> str:
if val is None:
return get_text("traffic_na")
@@ -171,7 +174,7 @@ async def my_subscription_command_handler(
traffic_limit=limit_display,
traffic_used=used_display,
traffic_left=remaining_display,
config_link=active.get("config_link") or get_text("config_link_not_available"),
config_link=config_link_value,
)
else:
text = get_text(
@@ -179,7 +182,7 @@ async def my_subscription_command_handler(
end_date=end_date.strftime("%Y-%m-%d") if end_date else "N/A",
days_left=max(0, days_left),
status=active.get("status_from_panel", get_text("status_active")).capitalize(),
config_link=active.get("config_link") or get_text("config_link_not_available"),
config_link=config_link_value,
traffic_limit=(f"{active['traffic_limit_bytes'] / 2**30:.2f} GB" if active.get("traffic_limit_bytes") else get_text("traffic_unlimited")),
traffic_used=(
f"{active['traffic_used_bytes'] / 2**30:.2f} GB" if active.get("traffic_used_bytes") is not None else get_text("traffic_na")
@@ -202,7 +205,7 @@ async def my_subscription_command_handler(
)
])
else:
cfg_link_val = (active or {}).get("config_link")
cfg_link_val = connect_button_url or config_link_display
if cfg_link_val:
prepend_rows.append([
InlineKeyboardButton(
bot/handlers/user/trial_handler.py
+21 -6
View File
@@ -13,6 +13,7 @@ from bot.keyboards.inline.user_keyboards import (
get_main_menu_inline_keyboard,
get_connect_and_main_keyboard,
)
from bot.utils.config_link import prepare_config_links
from bot.middlewares.i18n import JsonI18n
from .start import send_main_menu
@@ -76,7 +77,9 @@ async def request_trial_confirmation_handler(
final_message_text_in_chat = ""
show_trial_button_after_action = False
config_link_display_for_trial = None
config_link_for_trial = None
connect_button_url_for_trial = None
if activation_result and activation_result.get("activated"):
try:
@@ -85,9 +88,10 @@ async def request_trial_confirmation_handler(
pass
end_date_obj = activation_result.get("end_date")
config_link_for_trial = activation_result.get("subscription_url") or _(
"config_link_not_available"
config_link_display_for_trial, connect_button_url_for_trial = prepare_config_links(
settings, activation_result.get("subscription_url")
)
config_link_for_trial = config_link_display_for_trial or _("config_link_not_available")
traffic_gb_val = activation_result.get(
"traffic_gb", settings.TRIAL_TRAFFIC_LIMIT_GB
@@ -142,7 +146,11 @@ async def request_trial_confirmation_handler(
reply_markup = (
get_connect_and_main_keyboard(
current_lang, i18n, settings, config_link_for_trial
current_lang,
i18n,
settings,
config_link_display_for_trial,
connect_button_url=connect_button_url_for_trial,
)
if activation_result and activation_result.get("activated")
else get_main_menu_inline_keyboard(
@@ -221,7 +229,9 @@ async def confirm_activate_trial_handler(
final_message_text_in_chat = ""
show_trial_button_after_action = False
config_link_display_for_trial = None
config_link_for_trial = None
connect_button_url_for_trial = None
if activation_result and activation_result.get("activated"):
try:
@@ -230,9 +240,10 @@ async def confirm_activate_trial_handler(
pass
end_date_obj = activation_result.get("end_date")
config_link_for_trial = activation_result.get("subscription_url") or _(
"config_link_not_available"
config_link_display_for_trial, connect_button_url_for_trial = prepare_config_links(
settings, activation_result.get("subscription_url")
)
config_link_for_trial = config_link_display_for_trial or _("config_link_not_available")
traffic_gb_val = activation_result.get(
"traffic_gb", settings.TRIAL_TRAFFIC_LIMIT_GB
@@ -275,7 +286,11 @@ async def confirm_activate_trial_handler(
reply_markup = (
get_connect_and_main_keyboard(
current_lang, i18n, settings, config_link_for_trial
current_lang,
i18n,
settings,
config_link_display_for_trial,
connect_button_url=connect_button_url_for_trial,
)
if activation_result and activation_result.get("activated")
else get_main_menu_inline_keyboard(
bot/keyboards/inline/user_keyboards.py
+4 -2
View File
@@ -375,10 +375,12 @@ def get_connect_and_main_keyboard(
i18n_instance,
settings: Settings,
config_link: Optional[str],
connect_button_url: Optional[str] = None,
preserve_message: bool = False) -> InlineKeyboardMarkup:
"""Keyboard with a connect button and a back to main menu button."""
_ = lambda key, **kwargs: i18n_instance.gettext(lang, key, **kwargs)
builder = InlineKeyboardBuilder()
button_target = connect_button_url or config_link
if settings.SUBSCRIPTION_MINI_APP_URL:
builder.row(
@@ -387,9 +389,9 @@ def get_connect_and_main_keyboard(
web_app=WebAppInfo(url=settings.SUBSCRIPTION_MINI_APP_URL),
)
)
elif config_link:
elif button_target:
builder.row(
InlineKeyboardButton(text=_("connect_button"), url=config_link)
InlineKeyboardButton(text=_("connect_button"), url=button_target)
)
else:
builder.row(
bot/services/crypto_pay_service.py
+13 -5
View File
@@ -17,6 +17,7 @@ from bot.keyboards.inline.user_keyboards import get_connect_and_main_keyboard
from bot.services.notification_service import NotificationService
from db.dal import payment_dal, user_dal
from bot.utils.text_sanitizer import sanitize_display_name, username_for_display
from bot.utils.config_link import prepare_config_links
class CryptoPayService:
@@ -188,7 +189,9 @@ class CryptoPayService:
lang = db_user.language_code if db_user and db_user.language_code else settings.DEFAULT_LANGUAGE
_ = lambda k, **kw: i18n.gettext(lang, k, **kw)
config_link = activation.get("subscription_url") or _("config_link_not_available")
raw_config_link = activation.get("subscription_url") if activation else None
display_link, button_link = prepare_config_links(settings, raw_config_link)
config_link_text = display_link or _("config_link_not_available")
final_end = activation.get("end_date")
applied_days = 0
if referral_bonus and referral_bonus.get("referee_new_end_date"):
@@ -199,7 +202,7 @@ class CryptoPayService:
text = _("payment_successful_traffic_full",
traffic_gb=str(int(traffic_gb)) if float(traffic_gb).is_integer() else f"{traffic_gb:g}",
end_date=final_end.strftime('%Y-%m-%d') if final_end else "—",
config_link=config_link)
config_link=config_link_text)
elif applied_days:
inviter_name_display = _("friend_placeholder")
if db_user and db_user.referred_by_id:
@@ -216,15 +219,20 @@ class CryptoPayService:
bonus_days=applied_days,
final_end_date=final_end.strftime('%Y-%m-%d'),
inviter_name=inviter_name_display,
config_link=config_link)
config_link=config_link_text)
else:
text = _("payment_successful_full",
months=int(months),
end_date=final_end.strftime('%Y-%m-%d') if final_end else "—",
config_link=config_link)
config_link=config_link_text)
markup = get_connect_and_main_keyboard(
lang, i18n, settings, config_link, preserve_message=True
lang,
i18n,
settings,
display_link,
connect_button_url=button_link,
preserve_message=True,
)
try:
await bot.send_message(
bot/services/freekassa_service.py
+10 -11
View File
@@ -20,6 +20,7 @@ from bot.keyboards.inline.user_keyboards import get_connect_and_main_keyboard
from bot.services.notification_service import NotificationService
from db.dal import payment_dal, user_dal
from bot.utils.text_sanitizer import sanitize_display_name, username_for_display
from bot.utils.config_link import prepare_config_links
class FreeKassaService:
@@ -317,13 +318,12 @@ class FreeKassaService:
lang = db_user.language_code if db_user and db_user.language_code else self.settings.DEFAULT_LANGUAGE
_ = lambda k, **kw: self.i18n.gettext(lang, k, **kw) if self.i18n else k
config_link = None
final_end = None
raw_config_link = activation.get("subscription_url") if activation else None
config_link_display, connect_button_url = prepare_config_links(self.settings, raw_config_link)
config_link_text = config_link_display or _("config_link_not_available")
final_end = activation.get("end_date") if activation else None
months = payment.subscription_duration_months or 1
sale_mode = "traffic" if self.settings.traffic_sale_mode else "subscription"
if activation:
config_link = activation.get("subscription_url")
final_end = activation.get("end_date")
applied_days = 0
if referral_bonus and referral_bonus.get("referee_new_end_date"):
@@ -333,8 +333,6 @@ class FreeKassaService:
if not final_end and activation and activation.get("end_date"):
final_end = activation["end_date"]
if not config_link:
config_link = _("config_link_not_available")
if final_end:
end_date_str = final_end.strftime("%Y-%m-%d")
else:
@@ -346,7 +344,7 @@ class FreeKassaService:
text = _("payment_successful_traffic_full",
traffic_gb=traffic_label,
end_date=end_date_str if final_end else "",
config_link=config_link)
config_link=config_link_text)
elif applied_days:
inviter_name_display = _("friend_placeholder")
if db_user and db_user.referred_by_id:
@@ -364,14 +362,14 @@ class FreeKassaService:
bonus_days=applied_days,
final_end_date=end_date_str,
inviter_name=inviter_name_display,
config_link=config_link,
config_link=config_link_text,
)
else:
text = _(
"payment_successful_full",
months=months,
end_date=end_date_str,
config_link=config_link,
config_link=config_link_text,
)
if provider_payment_id:
order_info_text = _(
@@ -385,7 +383,8 @@ class FreeKassaService:
lang,
self.i18n,
self.settings,
config_link,
config_link_display,
connect_button_url=connect_button_url,
preserve_message=True,
)
try:
bot/services/platega_service.py
+10 -10
View File
@@ -15,6 +15,7 @@ from bot.keyboards.inline.user_keyboards import get_connect_and_main_keyboard
from bot.services.notification_service import NotificationService
from db.dal import payment_dal, user_dal
from bot.utils.text_sanitizer import sanitize_display_name, username_for_display
from bot.utils.config_link import prepare_config_links
class PlategaService:
@@ -213,11 +214,9 @@ class PlategaService:
lang = db_user.language_code if db_user and db_user.language_code else self.settings.DEFAULT_LANGUAGE
_ = lambda k, **kw: self.i18n.gettext(lang, k, **kw) if self.i18n else k
config_link = (
activation.get("subscription_url")
if activation
else None
) or _("config_link_not_available")
raw_config_link = activation.get("subscription_url") if activation else None
config_link_display, connect_button_url = prepare_config_links(self.settings, raw_config_link)
config_link_text = config_link_display or _("config_link_not_available")
final_end = activation.get("end_date") if activation else None
applied_days = 0
applied_promo_days = activation.get("applied_promo_bonus_days", 0) if activation else 0
@@ -233,7 +232,7 @@ class PlategaService:
"payment_successful_traffic_full",
traffic_gb=traffic_label,
end_date=final_end.strftime("%Y-%m-%d") if final_end else "",
config_link=config_link,
config_link=config_link_text,
)
elif applied_days:
inviter_name_display = _("friend_placeholder")
@@ -253,7 +252,7 @@ class PlategaService:
bonus_days=applied_days,
final_end_date=final_end.strftime("%Y-%m-%d") if final_end else "",
inviter_name=inviter_name_display,
config_link=config_link,
config_link=config_link_text,
)
elif applied_promo_days and final_end:
text = _(
@@ -261,21 +260,22 @@ class PlategaService:
months=payment_months,
bonus_days=applied_promo_days,
end_date=final_end.strftime("%Y-%m-%d"),
config_link=config_link,
config_link=config_link_text,
)
else:
text = _(
"payment_successful_full",
months=payment_months,
end_date=final_end.strftime("%Y-%m-%d") if final_end else "",
config_link=config_link,
config_link=config_link_text,
)
markup = get_connect_and_main_keyboard(
lang,
self.i18n,
self.settings,
config_link,
config_link_display,
connect_button_url=connect_button_url,
preserve_message=True,
)
try:
bot/services/severpay_service.py
+10 -10
View File
@@ -18,6 +18,7 @@ from bot.services.notification_service import NotificationService
from bot.keyboards.inline.user_keyboards import get_connect_and_main_keyboard
from db.dal import payment_dal, user_dal
from bot.utils.text_sanitizer import sanitize_display_name, username_for_display
from bot.utils.config_link import prepare_config_links
class SeverPayService:
@@ -231,11 +232,9 @@ class SeverPayService:
lang = db_user.language_code if db_user and db_user.language_code else self.settings.DEFAULT_LANGUAGE
_ = lambda k, **kw: self.i18n.gettext(lang, k, **kw) if self.i18n else k
config_link = (
activation.get("subscription_url")
if activation
else None
) or _("config_link_not_available")
raw_config_link = activation.get("subscription_url") if activation else None
config_link_display, connect_button_url = prepare_config_links(self.settings, raw_config_link)
config_link_text = config_link_display or _("config_link_not_available")
final_end = activation.get("end_date") if activation else None
applied_days = 0
applied_promo_days = activation.get("applied_promo_bonus_days", 0) if activation else 0
@@ -251,7 +250,7 @@ class SeverPayService:
"payment_successful_traffic_full",
traffic_gb=traffic_label,
end_date=final_end.strftime("%Y-%m-%d") if final_end else "",
config_link=config_link,
config_link=config_link_text,
)
elif applied_days:
inviter_name_display = _("friend_placeholder")
@@ -271,7 +270,7 @@ class SeverPayService:
bonus_days=applied_days,
final_end_date=final_end.strftime("%Y-%m-%d") if final_end else "",
inviter_name=inviter_name_display,
config_link=config_link,
config_link=config_link_text,
)
elif applied_promo_days and final_end:
text = _(
@@ -279,21 +278,22 @@ class SeverPayService:
months=payment_months,
bonus_days=applied_promo_days,
end_date=final_end.strftime("%Y-%m-%d"),
config_link=config_link,
config_link=config_link_text,
)
else:
text = _(
"payment_successful_full",
months=payment_months,
end_date=final_end.strftime("%Y-%m-%d") if final_end else "",
config_link=config_link,
config_link=config_link_text,
)
markup = get_connect_and_main_keyboard(
lang,
self.i18n,
self.settings,
config_link,
config_link_display,
connect_button_url=connect_button_url,
preserve_message=True,
)
try:
bot/services/stars_service.py
+13 -7
View File
@@ -13,6 +13,7 @@ from bot.middlewares.i18n import JsonI18n
from .notification_service import NotificationService
from bot.keyboards.inline.user_keyboards import get_connect_and_main_keyboard
from bot.utils.text_sanitizer import sanitize_display_name, username_for_display
from bot.utils.config_link import prepare_config_links
class StarsService:
@@ -121,16 +122,16 @@ class StarsService:
i18n: JsonI18n = i18n_data.get("i18n_instance")
_ = lambda k, **kw: i18n.gettext(current_lang, k, **kw) if i18n else k
config_link = activation_details.get("subscription_url") or _(
"config_link_not_available"
)
raw_config_link = activation_details.get("subscription_url") if activation_details else None
config_link_display, connect_button_url = prepare_config_links(self.settings, raw_config_link)
config_link_text = config_link_display or _("config_link_not_available")
if sale_mode == "traffic":
success_msg = _(
"payment_successful_traffic_full",
traffic_gb=str(int(months)) if float(months).is_integer() else f"{months:g}",
end_date=final_end.strftime('%Y-%m-%d'),
config_link=config_link,
config_link=config_link_text,
)
elif applied_days:
inviter_name_display = _("friend_placeholder")
@@ -150,17 +151,22 @@ class StarsService:
bonus_days=applied_days,
final_end_date=final_end.strftime('%Y-%m-%d'),
inviter_name=inviter_name_display,
config_link=config_link,
config_link=config_link_text,
)
else:
success_msg = _(
"payment_successful_full",
months=months,
end_date=final_end.strftime('%Y-%m-%d'),
config_link=config_link,
config_link=config_link_text,
)
markup = get_connect_and_main_keyboard(
current_lang, i18n, self.settings, config_link, preserve_message=True
current_lang,
i18n,
self.settings,
config_link_display,
connect_button_url=connect_button_url,
preserve_message=True,
)
try:
await self.bot.send_message(
bot/services/subscription_service.py
+5 -1
View File
@@ -7,6 +7,7 @@ from bot.middlewares.i18n import JsonI18n
from db.dal import user_dal, subscription_dal, promo_code_dal, payment_dal, user_billing_dal
from bot.utils.date_utils import add_months
from bot.utils.config_link import prepare_config_links
from db.models import User, Subscription
from config.settings import Settings
@@ -900,6 +901,8 @@ class SubscriptionService:
if panel_user_data.get("expireAt")
else None
)
config_link_raw = panel_user_data.get("subscriptionUrl")
display_link, connect_button_url = prepare_config_links(self.settings, config_link_raw)
hwid_limit = panel_user_data.get("hwidDeviceLimit")
if hwid_limit is None:
hwid_limit = self.settings.USER_HWID_DEVICE_LIMIT
@@ -908,7 +911,8 @@ class SubscriptionService:
"user_id": panel_user_data.get("uuid"),
"end_date": panel_end_date,
"status_from_panel": panel_user_data.get("status", "UNKNOWN").upper(),
"config_link": panel_user_data.get("subscriptionUrl"),
"config_link": display_link,
"connect_button_url": connect_button_url,
"traffic_limit_bytes": panel_user_data.get("trafficLimitBytes"),
"traffic_used_bytes": (panel_user_data.get("userTraffic") or {}).get("usedTrafficBytes"),
"user_bot_username": db_user.username,
bot/utils/config_link.py
+75
View File
@@ -0,0 +1,75 @@
import base64
import logging
from functools import lru_cache
from pathlib import Path
from typing import Optional, Tuple
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey
from config.settings import Settings
CRYPT4_PREFIX = "happ://crypt4/"
@lru_cache(maxsize=1)
def _load_crypt4_public_key() -> Optional[RSAPublicKey]:
"""Load and cache the happ crypt4 public key from the project root."""
pem_path = Path(__file__).resolve().parent.parent.parent / "happ-crypt4.pem"
try:
pem_bytes = pem_path.read_bytes()
key = serialization.load_pem_public_key(pem_bytes)
return key if isinstance(key, RSAPublicKey) else None
except FileNotFoundError:
logging.error("Crypt4 public key file not found at %s", pem_path)
except Exception as exc:
logging.error("Failed to load crypt4 public key: %s", exc, exc_info=True)
return None
def _encrypt_raw_link(raw_link: str) -> Optional[str]:
"""Encrypt the raw subscription URL with RSA PKCS#1 v1.5 and return base64 payload."""
public_key = _load_crypt4_public_key()
if not public_key:
return None
try:
encrypted = public_key.encrypt(raw_link.encode("utf-8"), padding.PKCS1v15())
return base64.urlsafe_b64encode(encrypted).decode("utf-8")
except Exception as exc:
logging.error("Failed to encrypt config link with crypt4: %s", exc, exc_info=True)
return None
def prepare_config_links(settings: Settings, raw_link: Optional[str]) -> Tuple[Optional[str], Optional[str]]:
"""
Build the user-facing connection key and the URL for the connect button.
Returns (display_link, button_link). When CRYPT4 is enabled the display link
is encrypted and prefixed with happ://crypt4/, and the button link is wrapped
with CRYPT4_REDIRECT_URL if provided.
"""
if not raw_link:
return None, None
cleaned = raw_link.strip()
if not cleaned:
return None, None
display_link = cleaned
button_link = cleaned
if settings.CRYPT4_ENABLED:
encrypted_payload = _encrypt_raw_link(cleaned)
if encrypted_payload:
display_link = f"{CRYPT4_PREFIX}{encrypted_payload}"
button_link = display_link
else:
logging.error("CRYPT4_ENABLED is set but encryption failed; using raw link as fallback.")
redirect_base = (settings.CRYPT4_REDIRECT_URL or "").strip()
if redirect_base and settings.CRYPT4_ENABLED and display_link:
button_link = f"{redirect_base}{display_link}"
return display_link, button_link
config/settings.py
+4
View File
@@ -164,6 +164,9 @@ class Settings(BaseSettings):
TRIAL_DURATION_DAYS: int = Field(default=3)
TRIAL_TRAFFIC_LIMIT_GB: Optional[float] = Field(default=5.0)
CRYPT4_ENABLED: bool = Field(default=False, description="Enable happ crypt4 encryption for subscription URLs")
CRYPT4_REDIRECT_URL: Optional[str] = Field(default=None, description="Base redirect URL used for the connect button when crypt4 is enabled")
WEB_SERVER_HOST: str = Field(default="0.0.0.0")
WEB_SERVER_PORT: int = Field(default=8080)
LOGS_PAGE_SIZE: int = Field(default=10)
@@ -499,6 +502,7 @@ class Settings(BaseSettings):
'PLATEGA_RETURN_URL',
'PLATEGA_FAILED_URL',
'SEVERPAY_RETURN_URL',
'CRYPT4_REDIRECT_URL',
mode='before',
)
@classmethod
happ-crypt4.pem
+14
View File
@@ -0,0 +1,14 @@
-----BEGIN PUBLIC KEY-----
MIICIjANBgkqhkiG9w0BAQEFAAOCAg8AMIICCgKCAgEA3UZ0M3L4K+WjM3vkbQnz
ozHg/cRbEXvQ6i4A8RVN4OM3rK9kU01FdjyoIgywve8OEKsFnVwERZAQZ1Trv60B
hmaM76QQEE+EUlIOL9EpwKWGtTL5lYC1sT9XJMNP3/CI0gP5wwQI88cY/xedpOEB
W72EmOOShHUm/b/3m+HPmqwc4ugKj5zWV5SyiT829aFA5DxSjmIIFBAms7DafmSq
LFTYIQL5cShDY2u+/sqyAw9yZIOoqW2TFIgIHhLPWek/ocDU7zyOrlu1E0SmcQQb
LFqHq02fsnH6IcqTv3N5Adb/CkZDDQ6HvQVBmqbKZKf7ZdXkqsc/Zw27xhG7OfXC
tUmWsiL7zA+KoTd3avyOh93Q9ju4UQsHthL3Gs4vECYOCS9dsXXSHEY/1ngU/hjO
WFF8QEE/rYV6nA4PTyUvo5RsctSQL/9DJX7XNh3zngvif8LsCN2MPvx6X+zLouBX
zgBkQ9DFfZAGLWf9TR7KVjZC/3NsuUCDoAOcpmN8pENBbeB0puiKMMWSvll36+2M
YR1Xs0MgT8Y9TwhE2+TnnTJOhzmHi/BxiUlY/w2E0s4ax9GHAmX0wyF4zeV7kDkc
vHuEdc0d7vDmdw0oqCqWj0Xwq86HfORu6tm1A8uRATjb4SzjTKclKuoElVAVa5Jo
oh/uZMozC65SmDw+N5p6Su8CAwEAAQ==
-----END PUBLIC KEY-----
requirements.txt
+1
View File
@@ -9,3 +9,4 @@ sqlalchemy[asyncio]==2.0.29
asyncpg==0.29.0
alembic==1.13.1
aiocryptopay==0.4.8
cryptography==42.0.8