"""Set or update a user's IANA timezone in Redis."""
from __future__ import annotations
import jsonutil as json
import logging
import re
import zoneinfo
from typing import Any, TYPE_CHECKING
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
TOOL_NO_BACKGROUND = True
TOOL_ALLOW_REPEAT = True
TOOL_NAME = "set_user_timezone"
TOOL_DESCRIPTION = (
"Set the active IANA timezone for a specific user ID. "
"Defaults to setting the invoking user's timezone. "
"Setting another user's timezone requires ALTER_PRIVILEGES."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"timezone": {
"type": "string",
"description": "Valid IANA timezone name (e.g. 'America/New_York', 'Europe/London', 'UTC').",
},
"user_id": {
"type": "string",
"description": "Optional target user ID (defaults to the invoking user). Changing other users requires ALTER_PRIVILEGES.",
},
},
"required": ["timezone"],
}
[docs]
async def run(
timezone: str,
user_id: str | None = None,
ctx: ToolContext | None = None,
**_kwargs,
) -> str:
"""Execute the tool to set user timezone in Redis.
Args:
timezone (str): Valid IANA timezone name.
user_id (str | None): Optional target user ID.
ctx (ToolContext | None): Tool context injection.
Returns:
str: JSON-serialized result.
"""
if ctx is None:
return json.dumps({"success": False, "error": "No tool context available."})
invoking_user_id = getattr(ctx, "user_id", None) or ""
if not invoking_user_id:
return json.dumps({"success": False, "error": "No invoking user ID found in context."})
# 1. Sanitization: Timezone Validation
if not timezone or not isinstance(timezone, str):
return json.dumps({"success": False, "error": "Invalid timezone parameter type."})
timezone_stripped = timezone.strip()
try:
zoneinfo.ZoneInfo(timezone_stripped)
except zoneinfo.ZoneInfoNotFoundError:
return json.dumps({
"success": False,
"error": f"Invalid IANA timezone name: '{timezone_stripped}'. Must be a valid name from standard IANA zoneinfo database (e.g., 'America/New_York', 'Europe/London', 'UTC')."
})
except Exception as e:
return json.dumps({
"success": False,
"error": f"Failed to validate timezone: {str(e)}"
})
# 2. Sanitization: User ID Sanitization
target_user_id = user_id if user_id is not None else invoking_user_id
if not isinstance(target_user_id, str):
return json.dumps({"success": False, "error": "Invalid user_id parameter type."})
target_user_id = target_user_id.strip()
# Strip any characters not in alphanumeric, _, -, :, ., or @ to prevent injection
sanitized_user_id = re.sub(r"[^a-zA-Z0-9_\-:.@]", "", target_user_id)
if not sanitized_user_id:
return json.dumps({"success": False, "error": "User ID cannot be empty after sanitization."})
if len(sanitized_user_id) > 128:
return json.dumps({"success": False, "error": "User ID is too long (maximum 128 characters)."})
# 3. Security/Privilege Gating
redis = getattr(ctx, "redis", None)
config = getattr(ctx, "config", None)
guild_id = getattr(ctx, "guild_id", "")
channel_id = getattr(ctx, "channel_id", "")
if sanitized_user_id != invoking_user_id:
from tools.alter_privileges import has_scoped_privilege, PRIVILEGES
has_priv = await has_scoped_privilege(
redis,
invoking_user_id,
PRIVILEGES["ALTER_PRIVILEGES"],
config,
guild_id=guild_id,
channel_id=channel_id,
)
if not has_priv:
logger.warning(
f"SECURITY: Unauthorized user {invoking_user_id} attempted to alter "
f"timezone for user {sanitized_user_id} — DENIED"
)
return json.dumps({
"success": False,
"error": "Unauthorized: You do not have privilege ALTER_PRIVILEGES to change other users' timezones."
})
# 4. Redis Update
if redis is None:
return json.dumps({"success": False, "error": "Redis client is not available in context."})
redis_key = f"stargazer:user:timezone:{sanitized_user_id}"
try:
await redis.set(redis_key, timezone_stripped)
logger.info(
f"Successfully updated timezone for user {sanitized_user_id} to "
f"'{timezone_stripped}' in Redis key '{redis_key}' by {invoking_user_id}."
)
return json.dumps({
"success": True,
"user_id": sanitized_user_id,
"timezone": timezone_stripped,
"msg": f"Successfully updated timezone to '{timezone_stripped}'."
})
except Exception as e:
logger.error(f"Failed to save timezone to Redis for user {sanitized_user_id}: {e}", exc_info=True)
return json.dumps({
"success": False,
"error": f"Failed to save timezone to Redis: {str(e)}"
})