首页
学习
活动
专区
工具
TVP
发布
社区首页 >问答首页 >Python Censys导出API

Python Censys导出API
EN

Stack Overflow用户
提问于 2018-08-01 02:55:26
回答 1查看 1.1K关注 0票数 0

我在弄清楚如何让Censys Python API将搜索查询导出到CSV文件时遇到了问题。显然,它有一个导出API,我尝试将它导入到我的代码中,但它仍然给我一个未定义的函数调用。

#!/user/bin/python3

#import sys
#import requests
#import os

import censys.ipv4
import censys.websites
import censys.query
import censys.export
#import censys

UID = "xxxxxx"
SECRET = "xxxxxx"

#res = requests.post(API_URL, auth=(UID, SECRET))
#if res.status_code != 200:
    #print ("error occurred!:", res.status_code)
    #sys.exit(1)

c = censys.ipv4.CensysIPv4(api_id = UID, api_secret = SECRET)
d = censys.export.CensysExport(api_id = UID, api_secret = SECRET)

fields_IPv4 = ["ip", "location.city", "location.country", "location.country_code", "location.postal_code", "autonomous_system.name", "autonomous_system.organization"]

for ipv4 in c.search("bah.com or boozallen", fields = fields_IPv4):
    try:
        print (ipv4["ip"])
    except:
        print ("N/A IP")
    try:
        print (ipv4["location.city"]) 
    except:
        print ("N/A City")
    try:
        print (ipv4["location.country"])
    except:
        print ("N/A Country") 
    try:
        print (ipv4["location.country_code"]) 
    except:
        print ("N/A  Country Code")
    try:
        print (ipv4["location.postal_code"]) 
    except:
        print ("N/A Postal Code")
    try:
        print (ipv4["autonomous_system.name"]) 
    except:
        print ("N/A System Name")
    try:
        print (ipv4["autonomous_system.organization"])
    except:
        print ("N/A System Organization")
    print ()

#NewJob = d.new_job("select count(*) from certificates.certificates")
#JobID = NewJob["job_id"]

censys_start_export(d, format = c("csv"), flatten = TRUE, compress = FALSE, delimiter = ",", headers = TRUE)

我可以得到我指定在Ubuntu 14.04的终端上显示的搜索查询信息,但在它完成显示后,我得到了这个错误:

压缩回溯(最近一次调用):文件"CensysAccumulationScript.py",第60行,格式为censys_start_export(d,

= c("csv"),flatten = TRUE,compress = FALSE,delimiter = ",",headers = TRUE) NameError:名称'censys_start_export‘未定义

我以为导入censys.export可以让我使用censys_start_export函数,但显然不是这样工作的。我尝试导入整个censys库,但也不起作用。

任何见解都将不胜感激。

EN

回答 1

Stack Overflow用户

发布于 2018-08-01 03:39:27

实际上,这里不需要censys.export.CensysExport类。您可以在最后阅读有关如何使用censys.export.CensysExport的内容。

import csv
from censys.ipv4 import CensysIPv4


UID = "<your-uid>"
SECRET = "<your-secret>"

ipv4 = CensysIPv4(api_id=UID, api_secret=SECRET)

fields_IPv4 = [
    "ip",
    "location.city",
    "location.country",
    "location.country_code",
    "location.postal_code",
    "autonomous_system.name",
    "autonomous_system.organization"
]

result = ipv4.search("bah.com or boozallen", fields=fields_IPv4)

censys.ipv4.CensysIPv4.search方法调用返回的结果是一个生成器对象。

您可以使用生成器产生的值(键为fields_IPv4的字典对象),并以这种方式将它们写入csv文件。这是可能的,因为csv.writer.writerows接受将迭代器传递给它。

with open('some.csv', 'w', newline='') as f:
    writer = csv.DictWriter(f, fields_IPv4)
    writer.writerows(result)

使用censys.export.CensysExport

这里的

censys.export.CensysExport的源代码

class CensysExport(CensysAPIBase):

    def new_job(self, query, format="json", flatten=False, compress=False,
                delimiter=None, headers=None):
        assert format in ("json", "csv")
        assert flatten in (True, False)
        assert compress in (True, False)
        data = {
            "query": query,
            "format": format,
            "flatten": flatten,
            "compress": compress,
            "delimiter": delimiter,
            "headers": headers
        }
        return self._post("export", data=data)

    def check_job(self, job_id):
        path = "/".join(("export", job_id))
        return self._get(path)

    def check_job_loop(self, job_id):
        while True:
            res = self.check_job(job_id)
            if res["status"] != "pending":
                return res
            time.sleep(1)

要使用它,您必须传递一个针对API运行的SQL查询。例如:

import censys.export import CensysExport

query = 'select * from certificates.certificates limit 1000' # shamelessly plucked from tests ;)

exporter = CensysExport(api_id=UID, api_secret=SECRET)
job_id = exporter.new_job(query, format="csv", delimiter=",", headers=True)
csv = exporter.check_job_loop(job_id)

with open('some.csv', 'w') as f:
    f.writelines(csv)

目前,这会引发一个CensysNotFoundException,因为没有导出端点,我在online API documentation中确认了这一点。

票数 2
EN
页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接:

https://stackoverflow.com/questions/51620248

复制
相关文章

相似问题

领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档