/usr/lib/imh-ultrastack-common/venv/lib/python3.13/site-packages/ngxutil
Edit: /usr/lib/imh-ultrastack-common/venv/lib/python3.13/site-packages/ngxutil/cache.py (6938B)
# vim: set ts=4 sw=4 expandtab syntax=python:
"""
ngxutil.cache
Nginx Cache Manager
@author J. Hipps
"""
import os
import re
import logging
import shutil
import struct
from urllib.parse import urlparse
from hashlib import md5
from errno import *
import requests
import arrow
from ngxutil import default_cache_base
from ngxutil.util import *
logger = logging.getLogger('ngxutil')
# Disable insecure warnings in newer versions of Requests module
if 'packages' in requests.__dict__:
# pylint: disable=no-member,no-name-in-module,import-error,wrong-import-position
from requests.packages.urllib3.exceptions import InsecureRequestWarning, InsecurePlatformWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
requests.packages.urllib3.disable_warnings(InsecurePlatformWarning)
def find_cache_item_url(zone, url, method='GET', cbase=default_cache_base):
"""
Check if a cache item exists by URL and method, then return its contents
"""
try:
upx = urlparse(url)
ukey = upx.scheme + method + upx.netloc + upx.path
except Exception as e:
logger.error("Failed to parse URL '%s': %s", url, str(e))
return None
return find_cache_item(zone, ukey, cbase)
def purge_cache_item(zone, key, cbase=default_cache_base):
"""
Purges an item from @zone cache under @key at basedir @cbase
Returns True on success, False on failure, None if no match
"""
zpath = os.path.join(cbase, zone)
chash = md5(key.encode('utf8', errors='ignore')).hexdigest()
cpath = os.path.join(zpath, chash[-1], chash)
logger.debug("find_cache_item: %s:%s --> %s", zone, key, cpath)
if os.path.exists(cpath):
try:
os.unlink(cpath)
except Exception as e:
logger.error("Failed to remove cache item [%s]: %s", cpath, str(e))
return False
logger.info("Purged cache entry: %s", cpath)
return True
else:
logger.debug("No cache entry exists at: %s", cpath)
return None
def find_cache_item(zone, key, cbase=default_cache_base):
"""
Check if a cache item exists, then return its contents
"""
zpath = os.path.join(cbase, zone)
chash = md5(key.encode('utf8', errors='ignore')).hexdigest()
cpath = os.path.join(zpath, chash[-1], chash)
logger.debug("find_cache_item: %s:%s --> %s", zone, key, cpath)
return read_cache_file(cpath, zone)
def read_cache_file(cpath, zone=None):
"""
Parse cache file located at @cpath
"""
try:
logger.debug("Reading cache item from file: %s", cpath)
with open(cpath, 'rb') as f:
craw = f.read()
cstat = os.fstat(f.fileno())
except IOError as e:
if e.errno == ENOENT:
return None
else:
logger.error("Failed to read cache file [%s]: %s", cpath, str(e))
return None
except Exception as e:
logger.error("Failed to read cache file [%s]: %s", cpath, str(e))
return None
# unpack cache header
hpack = struct.unpack('4Q', craw[:4*8])
expiry = hpack[1]
# parse cached response header
hhead, body = re.search(b'^(KEY:.+)\r\n\r\n(.*)$', craw, re.M|re.S).groups()
headers = {}
key = None
for thead in hhead.splitlines():
try:
tkey, tval = thead.split(b':', 1)
if tkey == b'KEY':
key = tval
else:
headers[tkey.decode('utf8', errors='ignore')] = tval.decode('utf8', errors='ignore')
except:
if thead.startswith(b'HTTP/'):
try:
status = thead.split(b' ', 1)[1]
except:
pass
# build cache object
tcache = {
'zone': zone,
'key': key.decode('utf8', errors='ignore').strip(),
'status': status.decode('utf8', errors='ignore'),
'expiry': expiry,
'expiry_txt': arrow.get(hpack[1]).humanize(),
'headers': headers,
'stat': cstat,
'expired': arrow.now() >= arrow.get(expiry),
'body': body.decode('utf8', errors='ignore'),
'path': cpath
}
return tcache
def purge_cache_zone(zone, cbase=default_cache_base):
"""
Purge cache zone @zone, located in @cbase
"""
zpath = os.path.join(cbase, zone)
try:
shutil.rmtree(zpath)
except Exception as e:
logger.error("Failed to remove cache zone [%s] at [%s]: %s", zone, zpath, str(e))
return False
logger.info("Purged cache zone [%s]", zone)
return True
def purge_full_cache(cbase=default_cache_base):
"""
Purges cache for ALL users
For safety, this method will only delete directories
that match users actually on the server.
"""
# Enumerate cPanel users
try:
ulist = os.listdir("/var/cpanel/users")
logger.debug("Enumerated %d users", len(ulist))
except Exception as e:
logger.error("Failed to enumerate users from /var/cpanel/users: %s", str(e))
return None
status = {'ok': 0, 'total': 0}
for tuser in ulist:
if tuser in ['system', 'root']:
continue
if purge_cache_zone(tuser, cbase):
status['ok'] += 1
status['total'] += 1
logger.info("Purged cache for %d/%d zones", status['ok'], status['total'])
return True
def purge_url(url):
"""
Purge @url. Prefixes /purge to the URI and makes a request
"""
uparse = urlparse(url)
ppath = uparse.path.strip()
if ppath == "":
ppath = "/"
purge_url = "%s://%s/purge%s" % (uparse.scheme, uparse.netloc, ppath)
try:
r = requests.get(purge_url, verify=False)
except Exception as e:
logger.error("Purge request [%s] failed: %s", purge_url, str(e))
return False
if r.status_code == 200 or r.status_code == 204:
logger.info("[%s] Purged page from cache successfully", url)
return True
elif r.status_code == 404:
logger.info("[%s] Page is not cached", url)
return False
else:
logger.error("[%s] Got an unexpected response (%d %s)", url, r.status_code, r.reason)
return False
def purge_url_all(url, user):
"""
Purge all related to @url (http, https, www, non-www) for @user
Used by the cache manager plugin
"""
uparse = urlparse(url)
ppath = uparse.path.strip()
if ppath == '':
ppath = '/'
if uparse.query:
ppath += '?' + uparse.query
# Iterate through each scheme
purged = 0
for tscheme in ('http', 'https'):
for twww in ('', 'www.'):
cache_key = "%s%s%s%s%s" % (tscheme, 'GET', twww, uparse.netloc, ppath)
if purge_cache_item(user, cache_key) is True:
purged += 1
return purged