Add basic auth ureqeusts and elementtree for xml.
Implement nextcloud directory listing.
This commit is contained in:
@@ -55,6 +55,12 @@ class Element:
|
||||
def get(self, key, default=None):
|
||||
return self.attrib.get(key, default)
|
||||
|
||||
def find_first_by_tag(self, tag):
|
||||
for child in self:
|
||||
if child.tag == tag:
|
||||
return child
|
||||
return None
|
||||
|
||||
def set(self, key, value):
|
||||
self.attrib[key] = value
|
||||
|
||||
|
||||
8
main.py
Normal file
8
main.py
Normal file
@@ -0,0 +1,8 @@
|
||||
import sdcard
|
||||
import uNextcloud
|
||||
import gc
|
||||
|
||||
sd_spi = machine.SPI(0, sck=machine.Pin(18, machine.Pin.OUT), mosi=machine.Pin(19, machine.Pin.OUT), miso=machine.Pin(16, machine.Pin.OUT))
|
||||
sd = sdcard.SDCard(sd_spi, machine.Pin(22))
|
||||
uos.mount(sd, "/sd")
|
||||
gc.collect()
|
||||
@@ -1,25 +1,51 @@
|
||||
import ElementTree
|
||||
from urllib import urequest
|
||||
import urequests
|
||||
from urequests import auth
|
||||
import gc
|
||||
import uos
|
||||
|
||||
|
||||
class uNextcloud:
|
||||
|
||||
def __init__(self):
|
||||
self.sd_spi = machine.SPI(0, sck=machine.Pin(18, machine.Pin.OUT), mosi=machine.Pin(19, machine.Pin.OUT), miso=machine.Pin(16, machine.Pin.OUT))
|
||||
self.sd = sdcard.SDCard(sd_spi, machine.Pin(22))
|
||||
class File:
|
||||
|
||||
def __init__(self, url_path, mimetype):
|
||||
self.url_path = url_path
|
||||
self.mimetype = mimetype
|
||||
|
||||
def get_url(self):
|
||||
return self.url_path
|
||||
|
||||
def get_mimetype(self):
|
||||
return self.mimetype
|
||||
|
||||
def __init__(self, nextcloud_url):
|
||||
self.url = nextcloud_url
|
||||
self.username = None
|
||||
self.password = None
|
||||
uos.mount(sd, "/sd")
|
||||
gc.collect()
|
||||
|
||||
def set_auth(self, username, password):
|
||||
self.username = username
|
||||
self.password = password
|
||||
|
||||
def get_folder_items(self, folder_path):
|
||||
pass
|
||||
def get_folder_items(self, folder_path=""):
|
||||
response = urequests.request("PROPFIND", self.url+"/remote.php/dav/files/"+self.username+"/"+folder_path, auth=urequests.auth.HTTPBasicAuth(self.username, self.password))
|
||||
if 200 <= response.status_code < 300:
|
||||
file_list = []
|
||||
try:
|
||||
xml = ElementTree.fromstring(response.text)
|
||||
for resp in xml:
|
||||
href = resp.find_first_by_tag("href")
|
||||
propstat = resp.find_first_by_tag("propstat")
|
||||
prop = propstat.find_first_by_tag("prop")
|
||||
content_type = prop.find_first_by_tag("getcontenttype")
|
||||
if content_type != None:
|
||||
file_list.append(self.File(href.text, content_type.text))
|
||||
return file_list
|
||||
except AttributeError:
|
||||
return None
|
||||
else:
|
||||
return None
|
||||
|
||||
def download_file_to_path(self, folder_path, file, destination_path):
|
||||
pass
|
||||
|
||||
169
urequests/__init__.py
Normal file
169
urequests/__init__.py
Normal file
@@ -0,0 +1,169 @@
|
||||
# (c) 2016-2021 Paul Sokolovsky, MIT license, https://github.com/pfalcon/pycopy-lib
|
||||
import usocket
|
||||
|
||||
class Request:
|
||||
pass
|
||||
|
||||
|
||||
class Response:
|
||||
|
||||
def __init__(self, f):
|
||||
self.raw = f
|
||||
self.encoding = "utf-8"
|
||||
self._cached = None
|
||||
|
||||
def close(self):
|
||||
if self.raw:
|
||||
self.raw.close()
|
||||
self.raw = None
|
||||
self._cached = None
|
||||
|
||||
@property
|
||||
def content(self):
|
||||
if self._cached is None:
|
||||
try:
|
||||
self._cached = self.raw.read()
|
||||
finally:
|
||||
self.raw.close()
|
||||
self.raw = None
|
||||
return self._cached
|
||||
|
||||
@property
|
||||
def text(self):
|
||||
return str(self.content, self.encoding)
|
||||
|
||||
def json(self):
|
||||
import ujson
|
||||
return ujson.loads(self.content)
|
||||
|
||||
|
||||
def request(method, url, data=None, json=None, headers={}, auth=None, stream=None, parse_headers=True):
|
||||
redir_cnt = 1
|
||||
if json is not None:
|
||||
assert data is None
|
||||
import ujson
|
||||
data = ujson.dumps(json)
|
||||
|
||||
while True:
|
||||
try:
|
||||
proto, dummy, host, path = url.split("/", 3)
|
||||
except ValueError:
|
||||
proto, dummy, host = url.split("/", 2)
|
||||
path = ""
|
||||
if proto == "http:":
|
||||
port = 80
|
||||
elif proto == "https:":
|
||||
import ussl
|
||||
port = 443
|
||||
else:
|
||||
raise ValueError("Unsupported protocol: " + proto)
|
||||
|
||||
if ":" in host:
|
||||
host, port = host.split(":", 1)
|
||||
port = int(port)
|
||||
|
||||
if auth is not None:
|
||||
req = Request()
|
||||
req.method = method
|
||||
req.url = url
|
||||
if not headers:
|
||||
# Fresh local dict, not a copy of anything.
|
||||
headers = {}
|
||||
req.headers = headers
|
||||
req = auth(req)
|
||||
headers = req.headers
|
||||
|
||||
ai = usocket.getaddrinfo(host, port, 0, usocket.SOCK_STREAM)
|
||||
ai = ai[0]
|
||||
|
||||
resp_d = None
|
||||
if parse_headers is not False:
|
||||
resp_d = {}
|
||||
|
||||
s = usocket.socket(ai[0], ai[1], ai[2])
|
||||
try:
|
||||
s.connect(ai[-1])
|
||||
if proto == "https:":
|
||||
s = ussl.wrap_socket(s, server_hostname=host)
|
||||
s.write(b"%s /%s HTTP/1.0\r\n" % (method, path))
|
||||
if not "Host" in headers:
|
||||
s.write(b"Host: %s\r\n" % host)
|
||||
# Iterate over keys to avoid tuple alloc
|
||||
for k in headers:
|
||||
s.write(k)
|
||||
s.write(b": ")
|
||||
s.write(headers[k])
|
||||
s.write(b"\r\n")
|
||||
if json is not None:
|
||||
s.write(b"Content-Type: application/json\r\n")
|
||||
if data:
|
||||
s.write(b"Content-Length: %d\r\n" % len(data))
|
||||
s.write(b"Connection: close\r\n\r\n")
|
||||
if data:
|
||||
s.write(data)
|
||||
|
||||
l = s.readline()
|
||||
#print(l)
|
||||
l = l.split(None, 2)
|
||||
status = int(l[1])
|
||||
reason = ""
|
||||
if len(l) > 2:
|
||||
reason = l[2].rstrip()
|
||||
while True:
|
||||
l = s.readline()
|
||||
if not l or l == b"\r\n":
|
||||
break
|
||||
#print(l)
|
||||
|
||||
if l.startswith(b"Transfer-Encoding:"):
|
||||
if b"chunked" in l:
|
||||
raise ValueError("Unsupported " + l.decode())
|
||||
elif l.startswith(b"Location:") and 300 <= status <= 399:
|
||||
if not redir_cnt:
|
||||
raise ValueError("Too many redirects")
|
||||
redir_cnt -= 1
|
||||
url = l[9:].decode().strip()
|
||||
#print("redir to:", url)
|
||||
status = 300
|
||||
break
|
||||
|
||||
if parse_headers is False:
|
||||
pass
|
||||
elif parse_headers is True:
|
||||
l = l.decode()
|
||||
k, v = l.split(":", 1)
|
||||
resp_d[k] = v.strip()
|
||||
else:
|
||||
parse_headers(l, resp_d)
|
||||
except OSError:
|
||||
s.close()
|
||||
raise
|
||||
|
||||
if status != 300:
|
||||
break
|
||||
|
||||
resp = Response(s)
|
||||
resp.status_code = status
|
||||
resp.reason = reason
|
||||
if resp_d is not None:
|
||||
resp.headers = resp_d
|
||||
return resp
|
||||
|
||||
|
||||
def head(url, **kw):
|
||||
return request("HEAD", url, **kw)
|
||||
|
||||
def get(url, **kw):
|
||||
return request("GET", url, **kw)
|
||||
|
||||
def post(url, **kw):
|
||||
return request("POST", url, **kw)
|
||||
|
||||
def put(url, **kw):
|
||||
return request("PUT", url, **kw)
|
||||
|
||||
def patch(url, **kw):
|
||||
return request("PATCH", url, **kw)
|
||||
|
||||
def delete(url, **kw):
|
||||
return request("DELETE", url, **kw)
|
||||
11
urequests/auth.py
Normal file
11
urequests/auth.py
Normal file
@@ -0,0 +1,11 @@
|
||||
# (c) 2021 Paul Sokolovsky, MIT license, https://github.com/pfalcon/pycopy-lib
|
||||
import uwwwauth
|
||||
|
||||
|
||||
class HTTPBasicAuth:
|
||||
def __init__(self, user, passwd):
|
||||
self.auth = uwwwauth.basic_resp(user, passwd)
|
||||
|
||||
def __call__(self, r):
|
||||
r.headers["Authorization"] = self.auth
|
||||
return r
|
||||
108
uwwwauth/__init__.py
Normal file
108
uwwwauth/__init__.py
Normal file
@@ -0,0 +1,108 @@
|
||||
# RFC2617, WWW-Authenticate: Basic/Digest module
|
||||
# (c) 2018 Paul Sokolovsky, MIT license
|
||||
import uhashlib
|
||||
import ubinascii
|
||||
|
||||
|
||||
# Private functions - do not use, will change
|
||||
|
||||
def md5_concat(arg1, arg2, arg3):
|
||||
h = uhashlib.md5(arg1)
|
||||
h.update(b":")
|
||||
h.update(arg2)
|
||||
if arg3 is not None:
|
||||
h.update(b":")
|
||||
h.update(arg3)
|
||||
return ubinascii.hexlify(h.digest()).decode()
|
||||
|
||||
|
||||
def make_digest_ha1(a1, method, uri, nonce):
|
||||
a2 = md5_concat(method, uri, None)
|
||||
digest = md5_concat(a1, nonce, a2)
|
||||
return digest
|
||||
|
||||
|
||||
def make_digest(realm, username, passwd, method, uri, nonce):
|
||||
a1 = md5_concat(username, realm, passwd)
|
||||
return make_digest_ha1(a1, method, uri, nonce)
|
||||
|
||||
|
||||
def parse_auth_req(line):
|
||||
typ, line = line.split(None, 1)
|
||||
d = {"type": typ}
|
||||
for kv in line.split(","):
|
||||
k, v = kv.split("=", 1)
|
||||
assert v[0] == '"' and v[-1] == '"'
|
||||
d[k.strip()] = v[1:-1]
|
||||
return d
|
||||
|
||||
|
||||
def format_resp(resp_d):
|
||||
fields = []
|
||||
for k, v in resp_d.items():
|
||||
if k in ("type", "passwd"):
|
||||
continue
|
||||
fields.append('%s="%s"' % (k, v))
|
||||
resp_auth = ", ".join(fields)
|
||||
|
||||
resp_auth = "Digest " + resp_auth
|
||||
return resp_auth
|
||||
|
||||
|
||||
def _digest_resp(auth_d, username, passwd, method, URL):
|
||||
#print(auth_d)
|
||||
|
||||
resp_d = {}
|
||||
resp_d["username"] = username
|
||||
resp_d["uri"] = URL
|
||||
resp_d["realm"] = auth_d["realm"]
|
||||
resp_d["nonce"] = auth_d["nonce"]
|
||||
|
||||
digest = make_digest(auth_d["realm"], username, passwd, method, URL, auth_d["nonce"])
|
||||
resp_d["response"] = digest
|
||||
#print(resp_d)
|
||||
|
||||
return format_resp(resp_d)
|
||||
|
||||
|
||||
# Helper functions - may change
|
||||
|
||||
def basic_resp(username, passwd):
|
||||
return "Basic " + ubinascii.b2a_base64("%s:%s" % (username, passwd))[:-1].decode()
|
||||
|
||||
|
||||
def auth_resp(auth_line, username, passwd, method=None, URL=None):
|
||||
auth_d = parse_auth_req(auth_line)
|
||||
if auth_d["type"] == "Basic":
|
||||
return basic_resp(username, passwd)
|
||||
elif auth_d["type"] == "Digest":
|
||||
assert method and URL
|
||||
return _digest_resp(auth_d, username, passwd, method, URL)
|
||||
else:
|
||||
raise ValueError(auth_d["type"])
|
||||
|
||||
|
||||
# Public interface
|
||||
|
||||
class WWWAuth:
|
||||
|
||||
def __init__(self, username, passwd):
|
||||
self.username = username
|
||||
self.passwd = passwd
|
||||
self.cached_auth_line = None
|
||||
|
||||
def resp(self, auth_line, method, URL):
|
||||
if auth_line.startswith("Basic"):
|
||||
return basic_resp(self.username, self.passwd)
|
||||
elif auth_line.startswith("Digest"):
|
||||
auth_d = parse_auth_req(auth_line)
|
||||
if auth_line != self.cached_auth_line:
|
||||
self.ha1 = md5_concat(self.username, auth_d["realm"], self.passwd)
|
||||
self.cached_auth_line = auth_line
|
||||
digest = make_digest_ha1(self.ha1, method, URL, auth_d["nonce"])
|
||||
auth_d["username"] = self.username
|
||||
auth_d["uri"] = URL
|
||||
auth_d["response"] = digest
|
||||
return format_resp(auth_d)
|
||||
else:
|
||||
raise ValueError("Unsupported auth: " + auth_line)
|
||||
Reference in New Issue
Block a user