Python pycurl 模块，HTTPAUTH 实例源码

我们从Python开源项目中，提取了以下6个代码示例，用于说明如何使用pycurl.HTTPAUTH。

项目：Mobotix-tools 作者：keptenkurk | 项目源码 | 文件源码

def transfer(ipaddr, username, password, commandfile):   
    #transfers commandfile to camera
    storage = StringIO()
    c = pycurl.Curl()
    c.setopt(c.URL, 'http://' + ipaddr + '/admin/remoteconfig')
    c.setopt(c.POST, 1)
    c.setopt(c.CONNECTTIMEOUT, 5)
    c.setopt(c.TIMEOUT, TIMEOUT)
    filesize = os.path.getsize(commandfile)
    f = open(commandfile, 'rb')
    c.setopt(c.FAILONERROR, True)
    c.setopt(pycurl.POSTFIELDSIZE, filesize)
    c.setopt(pycurl.READFUNCTION, FileReader(f).read_callback)
    c.setopt(c.WRITEFUNCTION, storage.write)
    c.setopt(pycurl.HTTPHEADER, ["application/x-www-form-urlencoded"])
    c.setopt(c.VERBOSE, VERBOSE)
    c.setopt(pycurl.HTTPAUTH, pycurl.HTTPAUTH_BASIC)
    c.setopt(pycurl.USERPWD, username + ':' + password)
    try:
        c.perform()
    except pycurl.error, error:
        errno, errstr = error
        print 'An error occurred: ', errstr
        return False, ''
    c.close()
    content = storage.getvalue()
    f.close()
    return True, content


# ***************************************************************
# *** Main program ***
# ***************************************************************

项目：Mobotix-tools 作者：keptenkurk | 项目源码 | 文件源码

def transfer(ipaddr, username, password, commandfile, trackprogress):   
    #transfers commandfile to camera
    storage = StringIO()
    c = pycurl.Curl()
    c.setopt(c.URL, 'http://' + ipaddr + '/admin/remoteconfig')
    c.setopt(c.POST, 1)
    c.setopt(c.CONNECTTIMEOUT, 5)
    c.setopt(c.TIMEOUT, TIMEOUT)
    filesize = os.path.getsize(commandfile)
    f = open(commandfile, 'rb')
    c.setopt(c.FAILONERROR, True)
    c.setopt(pycurl.POSTFIELDSIZE, filesize)
    c.setopt(pycurl.READFUNCTION, FileReader(f).read_callback)
    if trackprogress:
        c.setopt(pycurl.NOPROGRESS, 0)
        c.setopt(pycurl.PROGRESSFUNCTION, progresscallback)
        starttime = time.time()
    else:
        c.setopt(pycurl.NOPROGRESS, 1)  
    c.setopt(c.WRITEFUNCTION, storage.write)
    c.setopt(pycurl.HTTPHEADER, ["application/x-www-form-urlencoded"])
    c.setopt(c.VERBOSE, VERBOSE)
    c.setopt(pycurl.HTTPAUTH, pycurl.HTTPAUTH_BASIC)
    c.setopt(pycurl.USERPWD, username + ':' + password)
    try:
        c.perform()
    except pycurl.error, error:
        errno, errstr = error
        print 'An error occurred: ', errstr
        return False, ''
    c.close()
    content = storage.getvalue()
    f.close()
    return True, content

项目：Mobotix-tools 作者：keptenkurk | 项目源码 | 文件源码

def transfer(ipaddr, username, password, commandfile):   
    #transfers commandfile to camera
    storage = StringIO()
    c = pycurl.Curl()
    c.setopt(c.URL, 'http://' + ipaddr + '/admin/remoteconfig')
    c.setopt(c.POST, 1)
    c.setopt(c.CONNECTTIMEOUT, 5)
    c.setopt(c.TIMEOUT, 60)
    filesize = os.path.getsize(commandfile)
    f = open(commandfile, 'rb')
    c.setopt(c.FAILONERROR, True)
    c.setopt(pycurl.POSTFIELDSIZE, filesize)
    c.setopt(pycurl.READFUNCTION, FileReader(f).read_callback)
    c.setopt(c.WRITEFUNCTION, storage.write)
    c.setopt(pycurl.HTTPHEADER, ["application/x-www-form-urlencoded"])
    c.setopt(c.VERBOSE, 0)
    c.setopt(pycurl.HTTPAUTH, pycurl.HTTPAUTH_BASIC)
    c.setopt(pycurl.USERPWD, username + ':' + password)
    try:
        c.perform()
    except pycurl.error, error:
        errno, errstr = error
        print 'An error occurred: ', errstr
        return False, ''
    c.close()
    content = storage.getvalue()
    f.close()
    return True, content


# ***************************************************************
# *** Main program ***
# ***************************************************************

项目：darkc0de-old-stuff 作者：tuwid | 项目源码 | 文件源码

def perform(self):
        self.__performHead=""
        self.__performBody=""

        conn=pycurl.Curl()
        conn.setopt(pycurl.SSL_VERIFYPEER,False)
        conn.setopt(pycurl.SSL_VERIFYHOST,1)
        conn.setopt(pycurl.URL,self.completeUrl)

        if self.__method or self.__userpass:
            if self.__method=="basic":
                conn.setopt(pycurl.HTTPAUTH, pycurl.HTTPAUTH_BASIC)
            elif self.__method=="ntlm":
                conn.setopt(pycurl.HTTPAUTH, pycurl.HTTPAUTH_NTLM)
            elif self.__method=="digest":
                conn.setopt(pycurl.HTTPAUTH, pycurl.HTTPAUTH_DIGEST)
            conn.setopt(pycurl.USERPWD, self.__userpass)

        if self.__timeout:
            conn.setopt(pycurl.CONNECTTIMEOUT, self.__timeout)
            conn.setopt(pycurl.NOSIGNAL, 1)

        if self.__totaltimeout:
            conn.setopt(pycurl.TIMEOUT, self.__totaltimeout)
            conn.setopt(pycurl.NOSIGNAL, 1)

        conn.setopt(pycurl.WRITEFUNCTION, self.body_callback)
        conn.setopt(pycurl.HEADERFUNCTION, self.header_callback)

        if self.__proxy!=None:
            conn.setopt(pycurl.PROXY,self.__proxy)
            if self.__headers.has_key("Proxy-Connection"):
                del self.__headers["Proxy-Connection"]

        conn.setopt(pycurl.HTTPHEADER,self.__getHeaders())
        if self.method=="POST":
            conn.setopt(pycurl.POSTFIELDS,self.__postdata)
        conn.perform()

        rp=Response()
        rp.parseResponse(self.__performHead)
        rp.addContent(self.__performBody)

        self.response=rp

    ######### ESTE conjunto de funciones no es necesario para el uso habitual de la clase

项目：defcon-workshop 作者：devsecops | 项目源码 | 文件源码

def to_pycurl_object(c, req):

        c.setopt(pycurl.MAXREDIRS, 5)

        c.setopt(pycurl.WRITEFUNCTION, req.body_callback)
        c.setopt(pycurl.HEADERFUNCTION, req.header_callback)

        c.setopt(pycurl.NOSIGNAL, 1)
        c.setopt(pycurl.SSL_VERIFYPEER, False)
        c.setopt(pycurl.SSL_VERIFYHOST, 0)

        c.setopt(pycurl.URL,req.completeUrl)

        if req.getConnTimeout():
        c.setopt(pycurl.CONNECTTIMEOUT, req.getConnTimeout())

        if req.getTotalTimeout():
        c.setopt(pycurl.TIMEOUT, req.getTotalTimeout())


        authMethod, userpass = req.getAuth()
        if authMethod or userpass:
        if authMethod == "basic":
            c.setopt(pycurl.HTTPAUTH, pycurl.HTTPAUTH_BASIC)
        elif authMethod == "ntlm":
            c.setopt(pycurl.HTTPAUTH, pycurl.HTTPAUTH_NTLM)
        elif authMethod == "digest":
            c.setopt(pycurl.HTTPAUTH, pycurl.HTTPAUTH_DIGEST)
        c.setopt(pycurl.USERPWD, userpass)

        c.setopt(pycurl.HTTPHEADER, req.getHeaders())
        if req.method == "POST":
        c.setopt(pycurl.POSTFIELDS, req.postdata)

        if req.method != "GET" and req.method != "POST":
        c.setopt(pycurl.CUSTOMREQUEST, req.method)
        if req.method == "HEAD":
        c.setopt(pycurl.NOBODY, True)

        if req.followLocation:
        c.setopt(pycurl.FOLLOWLOCATION, 1)

        proxy = req.getProxy()
        if proxy != None:
            c.setopt(pycurl.PROXY, proxy)
            if req.proxytype=="SOCKS5":
                c.setopt(pycurl.PROXYTYPE,pycurl.PROXYTYPE_SOCKS5)
            elif req.proxytype=="SOCKS4":
                c.setopt(pycurl.PROXYTYPE,pycurl.PROXYTYPE_SOCKS4)
            req.delHeader("Proxy-Connection")

        return c

项目：wfuzz 作者：gwen001 | 项目源码 | 文件源码

def to_pycurl_object(c, req):

        c.setopt(pycurl.MAXREDIRS, 5)

        c.setopt(pycurl.WRITEFUNCTION, req.body_callback)
        c.setopt(pycurl.HEADERFUNCTION, req.header_callback)

        c.setopt(pycurl.NOSIGNAL, 1)
        c.setopt(pycurl.SSL_VERIFYPEER, False)
        c.setopt(pycurl.SSL_VERIFYHOST, 0)

        c.setopt(pycurl.URL,req.completeUrl)

        if req.getConnTimeout():
        c.setopt(pycurl.CONNECTTIMEOUT, req.getConnTimeout())

        if req.getTotalTimeout():
        c.setopt(pycurl.TIMEOUT, req.getTotalTimeout())


        authMethod, userpass = req.getAuth()
        if authMethod or userpass:
        if authMethod == "basic":
            c.setopt(pycurl.HTTPAUTH, pycurl.HTTPAUTH_BASIC)
        elif authMethod == "ntlm":
            c.setopt(pycurl.HTTPAUTH, pycurl.HTTPAUTH_NTLM)
        elif authMethod == "digest":
            c.setopt(pycurl.HTTPAUTH, pycurl.HTTPAUTH_DIGEST)
        c.setopt(pycurl.USERPWD, userpass)

        c.setopt(pycurl.HTTPHEADER, req.getHeaders())
        if req.method == "POST":
        c.setopt(pycurl.POSTFIELDS, req.postdata)

        if req.method != "GET" and req.method != "POST":
        c.setopt(pycurl.CUSTOMREQUEST, req.method)
        if req.method == "HEAD":
        c.setopt(pycurl.NOBODY, True)

        if req.followLocation:
        c.setopt(pycurl.FOLLOWLOCATION, 1)

        proxy = req.getProxy()
        if proxy != None:
            c.setopt(pycurl.PROXY, proxy)
            if req.proxytype=="SOCKS5":
                c.setopt(pycurl.PROXYTYPE,pycurl.PROXYTYPE_SOCKS5)
            elif req.proxytype=="SOCKS4":
                c.setopt(pycurl.PROXYTYPE,pycurl.PROXYTYPE_SOCKS4)
            req.delHeader("Proxy-Connection")

        return c