SRNI-2005/ctf

1from functools import lru_cache, wraps2from hashlib import md53from time import monotonic_ns4 5from flask import current_app, request6from flask_caching import Cache, make_template_fragment_key7 8 9class CTFdCache(Cache):10    """11    This subclass exists to give flask-caching some additional features12    Ideally likely we should have our own isolated redis connection but that might introduce more issues13    """14 15    def inc(self, *args, **kwargs):16        """17        Support redis INCR in flask-caching18        Note that redis INCR does not expire by default19        https://github.com/pallets-eco/flask-caching/issues/41820        """21        inc = getattr(self.cache, "inc", None)22        if inc is not None and callable(inc):23            return inc(*args, **kwargs)24        raise NotImplementedError25 26    def expire(self, key, timeout):27        """28        Support redis EXPIRE in flask-caching29        """30        if current_app.config["CACHE_TYPE"] == "redis":31            return self.cache._write_client.expire(32                f"{self.cache.key_prefix}{key}", timeout33            )34        else:35            # Generic alternative that leverages flask-caching built-ins to do expiration36            if timeout <= 0:37                self.cache.delete(key)38            value = self.get(key)39            if value:40                self.set(key=key, value=value, timeout=timeout)41                return True42            return False43 44 45cache = CTFdCache()46 47 48def timed_lru_cache(timeout: int = 300, maxsize: int = 64, typed: bool = False):49    """50    lru_cache implementation that includes a time based expiry51 52    Parameters:53    seconds (int): Timeout in seconds to clear the WHOLE cache, default = 5 minutes54    maxsize (int): Maximum Size of the Cache55    typed (bool): Same value of different type will be a different entry56 57    Implmentation from https://gist.github.com/Morreski/c1d08a3afa4040815eafd3891e16b945?permalink_comment_id=3437689#gistcomment-343768958    """59 60    def wrapper_cache(func):61        func = lru_cache(maxsize=maxsize, typed=typed)(func)62        func.delta = timeout * 10**963        func.expiration = monotonic_ns() + func.delta64 65        @wraps(func)66        def wrapped_func(*args, **kwargs):67            if monotonic_ns() >= func.expiration:68                func.cache_clear()69                func.expiration = monotonic_ns() + func.delta70            return func(*args, **kwargs)71 72        wrapped_func.cache_info = func.cache_info73        wrapped_func.cache_clear = func.cache_clear74        return wrapped_func75 76    return wrapper_cache77 78 79def make_cache_key(path=None, key_prefix="view/%s"):80    """81    This function mostly emulates Flask-Caching's `make_cache_key` function so we can delete cached api responses.82    Over time this function may be replaced with a cleaner custom cache implementation.83    :param path:84    :param key_prefix:85    :return:86    """87    if path is None:88        path = request.endpoint89    cache_key = key_prefix % path90    return cache_key91 92 93def make_cache_key_with_query_string(allowed_params=None, query_string_hash=None):94    if allowed_params is None:95        allowed_params = []96 97    def _make_cache_key_with_query_string(path=None, key_prefix="view/%s/%s"):98        if path is None:99            path = request.endpoint100 101        if query_string_hash:102            args_hash = query_string_hash103        else:104            args_hash = calculate_param_hash(105                params=tuple(request.args.items(multi=True)),106                allowed_params=allowed_params,107            )108        cache_key = key_prefix % (path, args_hash)109        return cache_key110 111    return _make_cache_key_with_query_string112 113 114def calculate_param_hash(params, allowed_params=None):115    # Copied from Flask-Caching but modified to allow only accepted parameters116    if allowed_params:117        args_as_sorted_tuple = tuple(118            sorted(pair for pair in params if pair[0] in allowed_params)119        )120    else:121        args_as_sorted_tuple = tuple(sorted(pair for pair in params))122    args_hash = md5(str(args_as_sorted_tuple).encode()).hexdigest()  # nosec B303 B324123    return args_hash124 125 126def clear_config():127    from CTFd.utils import _get_config, get_app_config128 129    cache.delete_memoized(_get_config)130    cache.delete_memoized(get_app_config)131 132 133def clear_standings():134    from CTFd.api import api135    from CTFd.api.v1.scoreboard import ScoreboardDetail, ScoreboardList136    from CTFd.constants.static import CacheKeys137    from CTFd.models import Teams, Users  # noqa: I001138    from CTFd.utils.scoreboard import get_scoreboard_detail139    from CTFd.utils.scores import get_standings, get_team_standings, get_user_standings140    from CTFd.utils.user import (141        get_team_place,142        get_team_score,143        get_user_place,144        get_user_score,145    )146 147    # Clear out the bulk standings functions148    cache.delete_memoized(get_standings)149    cache.delete_memoized(get_team_standings)150    cache.delete_memoized(get_user_standings)151    cache.delete_memoized(get_scoreboard_detail)152 153    # Clear out the individual helpers for accessing score via the model154    cache.delete_memoized(Users.get_score)155    cache.delete_memoized(Users.get_place)156    cache.delete_memoized(Teams.get_score)157    cache.delete_memoized(Teams.get_place)158 159    # Clear the Jinja Attrs constants160    cache.delete_memoized(get_user_score)161    cache.delete_memoized(get_user_place)162    cache.delete_memoized(get_team_score)163    cache.delete_memoized(get_team_place)164 165    # Clear out HTTP request responses166    cache.delete(make_cache_key(path=api.name + "." + ScoreboardList.endpoint))167    cache.delete(make_cache_key(path=api.name + "." + ScoreboardDetail.endpoint))168    cache.delete_memoized(ScoreboardList.get)169    cache.delete_memoized(ScoreboardDetail.get)170 171    # Clear out scoreboard templates172    cache.delete(make_template_fragment_key(CacheKeys.PUBLIC_SCOREBOARD_TABLE))173 174 175def clear_challenges():176    from CTFd.utils.challenges import get_all_challenges  # noqa: I001177    from CTFd.utils.challenges import (178        get_rating_average_for_challenge_id,179        get_solve_counts_for_challenges,180        get_solve_ids_for_user_id,181        get_solves_for_challenge_id,182        get_submissions_for_user_id_for_challenge_id,183    )184 185    cache.delete_memoized(get_all_challenges)186    cache.delete_memoized(get_solves_for_challenge_id)187    cache.delete_memoized(get_submissions_for_user_id_for_challenge_id)188    cache.delete_memoized(get_solve_ids_for_user_id)189    cache.delete_memoized(get_solve_counts_for_challenges)190    cache.delete_memoized(get_rating_average_for_challenge_id)191 192 193def clear_ratings():194    from CTFd.utils.challenges import get_rating_average_for_challenge_id195 196    cache.delete_memoized(get_rating_average_for_challenge_id)197 198 199def clear_pages():200    from CTFd.utils.config.pages import get_page, get_pages201 202    cache.delete_memoized(get_pages)203    cache.delete_memoized(get_page)204 205 206def clear_user_recent_ips(user_id):207    from CTFd.utils.user import get_user_recent_ips208 209    cache.delete_memoized(get_user_recent_ips, user_id=user_id)210 211 212def clear_user_session(user_id):213    from CTFd.utils.user import (  # noqa: I001214        get_user_attrs,215        get_user_place,216        get_user_recent_ips,217        get_user_score,218    )219 220    cache.delete_memoized(get_user_attrs, user_id=user_id)221    cache.delete_memoized(get_user_place, user_id=user_id)222    cache.delete_memoized(get_user_score, user_id=user_id)223    cache.delete_memoized(get_user_recent_ips, user_id=user_id)224 225 226def clear_all_user_sessions():227    from CTFd.utils.user import (  # noqa: I001228        get_user_attrs,229        get_user_place,230        get_user_recent_ips,231        get_user_score,232    )233 234    cache.delete_memoized(get_user_attrs)235    cache.delete_memoized(get_user_place)236    cache.delete_memoized(get_user_score)237    cache.delete_memoized(get_user_recent_ips)238 239 240def clear_team_session(team_id):241    from CTFd.utils.user import get_team_attrs, get_team_place, get_team_score242 243    cache.delete_memoized(get_team_attrs, team_id=team_id)244    cache.delete_memoized(get_team_place, team_id=team_id)245    cache.delete_memoized(get_team_score, team_id=team_id)246 247 248def clear_all_team_sessions():249    from CTFd.utils.user import get_team_attrs, get_team_place, get_team_score250 251    cache.delete_memoized(get_team_attrs)252    cache.delete_memoized(get_team_place)253    cache.delete_memoized(get_team_score)254