Repository: smicallef/spiderfoot
File: modules/sfp_zetalytics.py (Python)

Code before:

    def setup(self, sfc, userOpts=None):
        self.sf = sfc
        self.results = self.tempStorage()
        if userOpts:
            self.opts.update(userOpts)

    def watchedEvents(self):
        return ["INTERNET_NAME", "DOMAIN_NAME", "EMAILADDR"]

    def producedEvents(self):
        return ["INTERNET_NAME", "AFFILIATE_DOMAIN_NAME", "INTERNET_NAME_UNRESOLVED"]

    def emit(self, etype, data, pevent, notify=True):
        evt = SpiderFootEvent(etype, data, self.__name__, pevent)
        if notify:
            self.notifyListeners(evt)
        return evt

    def verify_emit_internet_name(self, hostname, pevent):
        if self.opts["verify"] and not self.sf.resolveHost(hostname):
            self.sf.debug(f"Host {hostname} could not be resolved")
            self.emit("INTERNET_NAME_UNRESOLVED", hostname, pevent)
        else:
            self.emit("INTERNET_NAME", hostname, pevent)

    def request(self, path, params):
        params = {**params, "token": self.opts["api_key"]}
        qs = urlencode(params)
        res = self.sf.fetchUrl(
            f"{self.BASE_URL}{path}/?{qs}",
            timeout=self.opts["_fetchtimeout"],
            useragent="SpiderFoot",
        )
        if res["content"] is None:
            self.sf.info(f"No Zetalytics info found for {path}?{qs}")
            return None
        try:
            info = json.loads(res["content"])
        except Exception as e:
            self.sf.error(
                f"Error processing JSON response from Zetalytics: {e}", False
            )
            return None

        return info

    def query_subdomains(self, domain):
        info = self.request("/subdomains", {"q": domain})
        return info


Review comment (refactor):
Rather than duplicate the `emit` function with a `verify_emit_internet_name` variant, I think from a design perspective it makes more sense to have the resolution logic outside and call the `emit` function with the appropriate arguments.


Code after:
        self.results = self.tempStorage()
        if userOpts:
            self.opts.update(userOpts)

    def watchedEvents(self):
        return ["INTERNET_NAME", "DOMAIN_NAME", "EMAILADDR"]

    def producedEvents(self):
        return ["INTERNET_NAME", "AFFILIATE_DOMAIN_NAME", "INTERNET_NAME_UNRESOLVED"]

    def emit(self, etype, data, pevent):
        if self.checkForStop():
            return None
        evt = SpiderFootEvent(etype, data, self.__name__, pevent)
        self.notifyListeners(evt)
        return evt

    def verify_emit_internet_name(self, hostname, pevent):
        if f"INTERNET_NAME:{hostname}" in self.results:
            return False
        if not self.getTarget().matches(hostname):
            return False
        if self.opts["verify"] and not self.sf.resolveHost(hostname):
            self.sf.debug(f"Host {hostname} could not be resolved")
            self.emit("INTERNET_NAME_UNRESOLVED", hostname, pevent)
            return True
        else:
            self.emit("INTERNET_NAME", hostname, pevent)
            if self.sf.isDomain(hostname, self.opts["_internettlds"]):
                self.emit("DOMAIN_NAME", hostname, pevent)
            return True

    def request(self, path, params):
        params = {**params, "token": self.opts["api_key"]}
        qs = urlencode(params)
        res = self.sf.fetchUrl(
            f"{self.BASE_URL}{path}/?{qs}",
            timeout=self.opts["_fetchtimeout"],
            useragent="SpiderFoot",
        )

        if res["content"] is None:
            self.sf.info(f"No Zetalytics info found for {path}?{qs}")
            return None

        try:
            return json.loads(res["content"])
        except Exception as e:
            self.sf.error(f"Error processing JSON response from Zetalytics: {e}")
        return None