RSS Feed

import feedparser
from bs4 import BeautifulSoup
from CommonServerPython import *

HTML_TAGS = ["p", "table", "ul", "ol", "h1", "h2", "h3", "h4", "h5", "h6"]

INTEGRATION_NAME = "RSS Feed"


class Client(BaseClient):
    """Client for RSS Feed - gets Reports from the website
    Attributes:
        server_url(str): The RSS URL.
        use_ssl: Whether to use ssl.
        proxy(str): Use system proxy.
    """

    def __init__(
        self,
        server_url,
        use_ssl,
        proxy,
        reliability,
        feed_tags,
        tlp_color,
        content_max_size=45,
        read_timeout=20,
        enrichment_excluded=False,
        headers=None,
    ):
        super().__init__(base_url=server_url, proxy=proxy, verify=use_ssl, headers=headers)
        self.feed_tags = feed_tags
        self.tlp_color = tlp_color
        self.content_max_size = content_max_size * 1000
        self.parsed_indicators = []
        self.feed_data = None
        self.reliability = reliability
        self.read_timeout = read_timeout
        self.enrichment_excluded = enrichment_excluded
        self.channel_link = None

    def request_feed_url(self):
        return self._http_request(method="GET", resp_type="response", timeout=self.read_timeout, full_url=self._base_url)

    def parse_feed_data(self, feed_response):
        try:
            if feed_response:
                self.feed_data = feedparser.parse(feed_response.text)
        except Exception as err:
            raise DemistoException(f"Failed to parse feed.\nError:\n{err!s}")

    def create_indicators_from_response(self):
        if hasattr(self.feed_data, "channel") and hasattr(self.feed_data.channel, "link"):  # type: ignore
            self.channel_link = self.feed_data.channel.link  # type: ignore
            if not self.channel_link.startswith(("http://", "https://")):
                self.channel_link = "https://" + self.channel_link
                demisto.debug(f"feed channel link is: {self.channel_link}")

        parsed_indicators: list = []
        if not self.feed_data:
            raise DemistoException(f"Could not parse feed data {self._base_url}")

        for indicator in reversed(self.feed_data.entries):
            link = indicator.get("link")
            if link and not link.startswith(("http://", "https://")):
                link = urljoin(self.channel_link, link)
                demisto.debug(f"indicator link is: {link}")

            publications = []
            if indicator:
                published = None
                if hasattr(indicator, "published"):
                    published = dateparser.parse(indicator.published)
                published_iso = published.strftime("%Y-%m-%dT%H:%M:%S") if published else ""

                publications.append(
                    {"timestamp": published_iso, "link": link, "source": self._base_url, "title": indicator.get("title")}
                )

                if indicator.get("links", []):
                    for tmp_link in indicator["links"]:
                        publications.append(
                            {
                                "timestamp": published_iso,
                                "link": tmp_link.href,
                                "source": self._base_url,
                                "title": indicator.get("title"),
                            }
                        )

                text = self.get_url_content(link)
                if not text:
                    continue
                indicator_obj = {
                    "type": "Report",
                    "value": indicator.get("title").replace(",", ""),  # Remove comma because the script of create
                    # relationship includes that as a list of titles.
                    "rawJSON": {"value": indicator, "type": "Report", "firstseenbysource": published_iso},
                    "reliability": self.reliability,
                    "fields": {
                        "rssfeedrawcontent": text,
                        "publications": publications,
                        "description": indicator.get("summary"),
                        "tags": self.feed_tags,
                    },
                }
                if self.tlp_color:
                    indicator_obj["fields"]["trafficlightprotocol"] = self.tlp_color

                if self.enrichment_excluded:
                    indicator_obj["enrichmentExcluded"] = self.enrichment_excluded

            parsed_indicators.append(indicator_obj)

        return parsed_indicators

    def get_url_content(self, link: str) -> str:
        """Returns the link content only from the relevant tags (listed on HTML_TAGS). For better performance - if the
        extracted content is bigger than "content_max_size" we trim him"""
        demisto.debug(f"Getting url content for {link=}")
        try:
            response_url = self._http_request(method="GET", full_url=link, resp_type="str", timeout=self.read_timeout)
            demisto.debug(f"Got response {response_url=}")
        except DemistoException as e:
            demisto.debug(f"Failed to get content for {link=} - Skipping\nError:\n{e}")
            return ""
        report_content = (
            "This is a dumped content of the article. Use the link under Publications field to read the full article. \n\n"
        )
        soup = BeautifulSoup(response_url.content, "html.parser")
        for tag in soup.find_all():
            if tag.name in HTML_TAGS:
                for string in tag.stripped_strings:
                    report_content += " " + string
        try:
            encoded_content = report_content.encode("utf-8", errors="replace")
        except Exception as err:
            demisto.debug(f"Fail encoding the article content, skipping report {link}. \nError:\n{err!s}")
            return ""
        if len(encoded_content) > self.content_max_size:  # Ensure report_content does not exceed the
            # indicator size limit (~50KB)
            report_content = encoded_content[: self.content_max_size].decode("utf-8", errors="replace")
            report_content += " This is truncated text, report content was too big."

        return report_content


def fetch_indicators(client: Client):
    feed_response = client.request_feed_url()
    client.parse_feed_data(feed_response)
    parsed_indicators = client.create_indicators_from_response()
    return parsed_indicators


def get_indicators(client: Client, indicators: list, args: dict) -> CommandResults:
    limit = int(args.get("limit", 10))
    parsed_indicators = indicators[:limit]
    parsed_for_hr = []
    for indicator in parsed_indicators:
        link = indicator.get("fields", {}).get("publications", [{}])[0].get("link")
        article_title = indicator.get("value", "")
        article_field_hr = article_title
        if link:
            article_field_hr = f"[{article_title}]({link})"  # if there is a link to the article, we want the
            # article's title to be a link.
        parsed_for_hr.append({"Article": article_field_hr, "Type": indicator.get("type")})
    headers = ["Article", "Type"]
    hr_ = tableToMarkdown(name=INTEGRATION_NAME, t=parsed_for_hr, headers=headers)
    return CommandResults(readable_output=hr_, raw_response=client.feed_data)


def check_feed(client: Client) -> str:
    feed_response = client.request_feed_url()
    if feed_response and "html" in feed_response.headers["content-type"]:
        raise DemistoException(
            f"{feed_response.url} is not rss feed url. Try look for a url containing xml format data,"
            f" that could be found under urls with 'feed' prefix or suffix."
        )
    else:
        client.parse_feed_data(feed_response)  # If parse response will raise an error, test will fail.
        return "ok"


def main():
    params = demisto.params()
    server_url = (params.get("server_url")).rstrip()
    default_headers = params.get("default_headers")
    if default_headers:
        try:
            default_headers = json.loads(default_headers.strip())
        except ValueError as e:
            return_error("Unable to parse Request headers value. Please verify the headers value is a valid JSON. - " + str(e))

    command = demisto.command()
    demisto.info(f"Command being called is {command}")

    try:
        reliability = params.get("feedReliability")
        reliability = reliability if reliability else DBotScoreReliability.F

        if DBotScoreReliability.is_valid_type(reliability):
            reliability = DBotScoreReliability.get_dbot_score_reliability_from_str(reliability)
        else:
            raise Exception("Please provide a valid value for the Source Reliability parameter.")
        client = Client(
            server_url=server_url,
            use_ssl=not params.get("insecure", False),
            proxy=params.get("proxy"),
            reliability=reliability,
            feed_tags=argToList(params.get("feedTags")),
            tlp_color=params.get("tlp_color"),
            content_max_size=int(params.get("max_size", "45")),
            read_timeout=int(params.get("read_timeout", "20")),
            enrichment_excluded=(
                argToBoolean(params.get("enrichmentExcluded", False))
                or (params.get("tlp_color") == "RED" and is_xsiam_or_xsoar_saas())
            ),
            headers=default_headers,
        )

        if command == "test-module":
            return_results(check_feed(client))

        elif command == "rss-get-indicators":
            parsed_indicators = fetch_indicators(client)
            return_results(get_indicators(client, parsed_indicators, demisto.args()))

        elif command == "fetch-indicators":
            parsed_indicators = fetch_indicators(client)
            for iter_ in batch(parsed_indicators, batch_size=2000):
                demisto.createIndicators(iter_)
        else:
            raise NotImplementedError(f"Command {command} is not implemented.")

    except Exception as err:
        return_error(f"Failed to execute {INTEGRATION_NAME} with {command} command.\nError:\n{err!s}")


if __name__ in ("builtin__", "builtins", "__main__"):
    main()