Page 4 of 4

Re: urequests and SSL on WiPy

Posted: Tue Sep 06, 2016 9:15 pm
by danielm
I think that there can only be one CA cert at a time. Maybe you could dynamically copy different ca.pem files to "/flash/cert" based on your needs during runtime of your application.

Btw, did you try if server cert validation works if you use different name than ca.pem and provide real name in ca_certs parameter of ussl.wrap_socket()?

Re: urequests and SSL on WiPy

Posted: Wed Sep 07, 2016 7:27 am
by jgmdavies

I haven't tried that experiment since things started working, because (i) earlier posts stated it has to be ca.pem, and (ii) when I copied other files previously, I found that it's only with ca.pem that I see the file magically 'disappear' when I FTP it (as it's presumably moved to internal memory as documented - actually, it never appears in the destination folder listing in my FTP app).


Re: urequests and SSL on WiPy

Posted: Wed Sep 07, 2016 8:49 am
by danielm
No problem, I will do the test and will post results here.

Btw I did similar modifications to umqtt.simple module and successfuly tested server(broker) certificate validation.

Re: urequests and SSL on WiPy

Posted: Sat Aug 22, 2020 1:04 pm
by Saran
jgmdavies wrote:
Tue Sep 06, 2016 12:25 pm

OK, I may have got it to work - there's conflicting info out there in Google land... This was a helpful post: viewtopic.php?t=1089

The cert file has to be in DER format, but must be named ca.pem. If you FTP it to /flash/cert it will 'disappear', which is expected as it's automatically moved to internal storage.

Here's my test code:

Code: Select all

import machine
from machine import RTC
import urequests

rtc = RTC(datetime=(2016, 9, 6, 9, 0, 0, 0, None)) # init with a specific time and date

url = ""
resp = urequests.get(url)

print("Response from GET to: " + url + ", status code " + str(resp.status_code))
Note that the RTC must be set for server certificate validation to work.

And here's my version of

Code: Select all


# From:
# with changes by J.G.Davies, Jacobus Systems Ltd.

import usocket

class Response:

	def __init__(self, f):
		self.raw = f
		self.encoding = "utf-8"
		self._cached = None

	def close(self):
		if self.raw:
			self.raw = None
		self._cached = None

	def content(self):
		if self._cached is None:
			self._cached =
			self.raw = None
		return self._cached

	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={}, stream=None):
		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
		raise ValueError("Unsupported protocol: " + proto)

	if ":" in host:
		host, port = host.split(":", 1)
		port = int(port)

	ai = usocket.getaddrinfo(host, port)
	addr = ai[0][4]

	# JGD 5 Sep 2016
	if proto == "https:":
		s = usocket.socket(usocket.AF_INET, usocket.SOCK_STREAM, usocket.IPPROTO_SEC)
		s = usocket.socket()

	if proto == "https:":
		# JGD 6 Sep 2016
		#s = ussl.wrap_socket(s)
		s = ussl.wrap_socket(s, keyfile=None, certfile=None, server_side=False, \
			cert_reqs=ussl.CERT_REQUIRED, ca_certs='/flash/cert/ca.pem')

	# JGD 6 Sep 2016 - Connect *after* the wrap.

	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(b": ")
	if json is not None:
		assert data is None
		import ujson
		data = ujson.dumps(json)
	if data:
		s.write(b"Content-Length: %d\r\n" % len(data))
	if data:

	l = s.readline()
	protover, status, msg = l.split(None, 2)
	status = int(status)

	while True:
		l = s.readline()
		if not l or l == b"\r\n":
		if l.startswith(b"Transfer-Encoding:"):
			if b"chunked" in line:
				raise ValueError("Unsupported " + l)
		elif l.startswith(b"Location:"):
			raise NotImplementedError("Redirects not yet supported")

	resp = Response(s)
	resp.status_code = status
	resp.reason = msg.rstrip()
	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)
Note the changes to socket creation and wrapping, and the connect() after the wrapping.

This works with status 200 on (then the previous memory allocation error if I try to get the content). With a junk file substituted for ca.pem, I get OSError 456.

I'm happy to try other test methods if anyone wants.

Hey Jim,

I'm using your version of urequests and with your code in REPL I get this error:

Code: Select all

>>> import machine
>>> from machine import RTC
>>> import urequests
>>> rtc = RTC(datetime=(2016, 9, 6, 9, 0, 0, 0, None)) # init with a specific time and date
>>> print(
(2016, 9, 6, 9, 0, 0, 61000, None)
>>> url = ""
>>> resp = urequests.get(url)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/flash/lib/", line 121, in get
I put the ca.pem from this link ... m?raw=true and put in in /flash/cert and rebooted.

My FW: MicroPython v1.12 on 2019-12-20; WiPy with CC3200

Any ideas?