server/plugins/defuzzer.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import calendar
import re
import shlex
import typing

"""
This is the query de-fuzzer library for Foal.
It turns a URL search query into an ES query

"""


def defuzz(formdata: dict, nodate: bool = False, list_override: typing.Optional[str] = None) -> dict:
    # Default to 30 day date range
    daterange = {"gt": "now-30d", "lt": "now+1d"}

    # Custom date range?
    # If a month is the only thing, fake start and end
    if "date" in formdata and "e" not in formdata:
        formdata["s"] = formdata["date"]
        formdata["e"] = formdata["date"]
    # classic start and end month params
    if "s" in formdata and "e" in formdata:
        if not re.match(r"\d{4}-\d{1,2}$", formdata["s"]):
            raise ValueError("Keyword 's' must be of type YYYY-MM")
        if not re.match(r"\d{4}-\d{1,2}$", formdata["e"]):
            raise ValueError("Keyword 'e' must be of type YYYY-MM")
        syear, smonth = formdata["s"].split("-", 1)
        eyear, emonth = formdata["e"].split("-", 1)
        _estart, eend = calendar.monthrange(int(eyear), int(emonth))
        daterange = {
            "gt": "%04u/%02u/01 00:00:00" % (int(syear), int(smonth)),
            "lt": "%04u/%02u/%02u 23:59:59" % (int(eyear), int(emonth), eend),
        }
    # days ago to start, and number of days to match
    elif "dfrom" in formdata and "dto" in formdata:
        dfrom = formdata["dfrom"]
        dto = formdata["dto"]
        if re.match(r"\d+$", dfrom) and re.match(r"\d+$", dto):
            ef = int(dfrom)
            et = int(dto)
            if ef > 0 and et > 0:
                if et > ef:
                    et = ef # avoid overruning into the future
                daterange = { 
                    "gte": "now-%dd" % ef,
                    "lte": "now-%dd" % (ef - et),
                }
        else:
            raise ValueError("Keywords 'dfrom' and 'dto' must be numeric")

    # Advanced date formatting
    elif "d" in formdata:
        # The more/less than N days/weeks/months/years ago
        m = re.match(r"^(lte|gte)=([0-9]+[Mwyd])$", formdata["d"])
        if m:
            t = m.group(1)
            r = m.group(2)
            if t == "lte" and r:
                daterange = {"gt": "now-%s" % r}
            elif t == "gte" and r:
                daterange = {"lt": "now-%s" % r}
        else:
            # simple one month listing
            m = re.match(r"^(\d\d\d\d-\d+)$", formdata["d"])
            if m:
                xdate = m.group(1)
                dyear, dmonth = xdate.split("-", 1)
                daterange = {
                    "gte": "%04u-%02u-01||/M" % (int(dyear), int(dmonth)),
                    "lte": "%04u-%02u-01||/M" % (int(dyear), int(dmonth)),
                    "format": "yyyy-MM-dd",
                }
            else:
                # dfr and dto defining a time span
                m = re.match(r"^dfr=(\d\d\d\d-\d+-\d+)\|dto=(\d\d\d\d-\d+-\d+)$", formdata["d"])
                if m:
                    dfr = m.group(1)
                    dto = m.group(2)
                    syear, smonth, sday = dfr.split("-", 2)
                    eyear, emonth, eday = dto.split("-", 2)
                    daterange = {
                        "gt": "%04u/%02u/%02u 00:00:00" % (int(syear), int(smonth), int(sday)),
                        "lt": "%04u/%02u/%02u 23:59:59" % (int(eyear), int(emonth), int(eday)),
                    }

    # List parameter(s)
    if list_override:  # Certain requests use the full list ID as a single variable. Allow for that if so.
        if not list_override.count("@") == 1:
            raise ValueError("list_override must contain exactly one @ character")
        listname, fqdn = list_override.split("@", 1)
    else:
        fqdn = formdata.get("domain", '')  # Must be provided
        listname = formdata.get("list", '')  # Must be provided
    if not fqdn:
        raise ValueError("You must specify a domain part of the mailing list(s) to search, or * for wildcard search.")
    if not listname:
        raise ValueError("You must specify a list part of the mailing list(s) to search, or * for wildcard search.")
    if "@" in listname:
        raise ValueError("The list component of the List ID(s) cannot contain @, please use both list and domain keywords for searching.")
    list_raw = "<%s.%s>" % (listname, fqdn)

    # Default is to look in a specific list
    query_list_hash: typing.Dict = {"term": {"list_raw": list_raw}}

    # *@fqdn match?
    if listname == "*" and fqdn != "*":
        query_list_hash = {"wildcard": {"list_raw": {"value": "*.%s>" % fqdn}}}

    # listname@* match?
    if listname != "*" and fqdn == "*":
        query_list_hash = {"wildcard": {"list_raw": "<%s.*>" % listname}}

    # *@* ??
    if listname == "*" and fqdn == "*":
        query_list_hash = {"wildcard": {"list_raw": "*"}}

    must = [query_list_hash]
    must_not = []

    # Append date range if not excluded
    if not nodate:
        must.append({"range": {"date": daterange}})

    # Query string search:
    # - foo bar baz: find emails with these words
    # - orange -apples: fond email with oranges but not apples
    # - "this sentence": find emails with this exact string
    if "q" in formdata:
        qs = formdata["q"].replace(":", "")
        try:
            bits = shlex.split(qs)
        except ValueError:  # Uneven number of quotes, default to split on whitespace instead
            bits = qs.split()

        query_should_match = []
        query_should_not_match = []

        for bit in bits:
            force_positive = False
            # Translate -- into a positive '-', so you can find "-1" etc
            if bit[0:2] == "--":
                force_positive = True
                bit = bit[1:]
            # Negatives
            if bit[0] == "-" and not force_positive:
                query_should_not_match.append(bit[1:])
            # Positives
            else:
                query_should_match.append(bit)

        if query_should_match:
            query_should_match_expanded = []
            for x in query_should_match:
                query_should_match_expanded.append(
                    {
                        "bool": {
                            "should": [
                                {
                                    "multi_match": {
                                        "fields": ["from", "body", "subject"],
                                        "query": x,
                                        "type": "phrase",
                                    },
                                },
                            ]
                        }
                    }
                )
            xmust = {"bool": {"minimum_should_match": len(query_should_match), "should": query_should_match_expanded}}
            must.append(xmust)

        for x in query_should_not_match:
            must_not.append(
                {
                    "match": {
                        "subject": x,
                    }
                }
            )
            must_not.append(
                {
                    "match": {
                        "from": x,
                    }
                }
            )
            must_not.append(
                {
                    "match": {
                        "body": x,
                    }
                }
            )

    # Header parameters
    for header in ["from", "subject", "body", "to", "messageid"]:
        hname = "header_%s" % header
        if hname in formdata:
            hvalue = formdata[hname]
            # '-' not allowed in variable names, so we convert here
            if header == 'messageid':
                header = 'message-id'
            must.append({"match_phrase": {header: hvalue}})

    query_as_bool = {"must": must}

    if must_not:
        query_as_bool["must_not"] = must_not

    return query_as_bool