add support for client TLS certificates

ircrobots/__init__.py

@@ -1,5 +1,5 @@
 from .bot    import Bot
 from .server import Server
-from .params import (ConnectionParams, SASLUserPass, SASLExternal, SASLSCRAM,
+from .params import (ConnectionParams, ClientTLSCertificate, SASLUserPass, SASLExternal, SASLSCRAM,
     STSPolicy, ResumePolicy)
 from .ircv3  import Capability

ircrobots/interface.py

@@ -5,7 +5,7 @@ from enum    import IntEnum
 from ircstates import Server, Emit
 from irctokens import Line, Hostmask
 
-from .params   import ConnectionParams, SASLParams, STSPolicy, ResumePolicy
+from .params   import ConnectionParams, SASLParams, STSPolicy, ResumePolicy, ClientTLSCertificate
 
 class ITCPReader(object):
     async def read(self, byte_count: int):
@@ -28,6 +28,7 @@ class ITCPTransport(object):
             port:        int,
             tls:         bool,
             tls_verify:  bool=True,
+            certificate: Optional[ClientTLSCertificate]=None,
             bindhost:    Optional[str]=None
             ) -> Tuple[ITCPReader, ITCPWriter]:
         pass

ircrobots/params.py

@@ -28,6 +28,12 @@ class ResumePolicy(object):
     address: str
     token:   str
 
+@dataclass
+class ClientTLSCertificate(object):
+    certfile: str
+    keyfile: Optional[str] = None
+    password: Optional[str] = None
+
 @dataclass
 class ConnectionParams(object):
     nickname: str
@@ -42,6 +48,7 @@ class ConnectionParams(object):
     password:    Optional[str] = None
     tls_verify:  bool = True
     sasl:        Optional[SASLParams] = None
+    certificate: Optional[ClientTLSCertificate] = None
 
     sts:    Optional[STSPolicy]    = None
     resume: Optional[ResumePolicy] = None

ircrobots/security.py

@@ -1,6 +1,13 @@
 import ssl
 
-def tls_context(verify: bool=True) -> ssl.SSLContext:
+from typing  import Optional
+from .params import ClientTLSCertificate
+
+def tls_context(
+    verify: bool=True,
+    certificate: Optional[ClientTLSCertificate]=None
+    ) -> ssl.SSLContext:
+
     context = ssl.SSLContext(ssl.PROTOCOL_TLS)
     context.options |= ssl.OP_NO_SSLv2
     context.options |= ssl.OP_NO_SSLv3
@@ -10,4 +17,11 @@ def tls_context(verify: bool=True) -> ssl.SSLContext:
     if verify:
         context.verify_mode = ssl.CERT_REQUIRED
 
+    if certificate is not None:
+        context.load_cert_chain(
+            certificate.certfile,
+            certificate.keyfile,
+            certificate.password
+        )
+
     return context

ircrobots/server.py

@@ -126,6 +126,7 @@ class Server(IServer):
             params.port,
             tls        =params.tls,
             tls_verify =params.tls_verify,
+            certificate=params.certificate,
             bindhost   =params.bindhost)
 
         self._reader = reader

ircrobots/transport.py

@@ -4,6 +4,7 @@ from asyncio       import StreamReader, StreamWriter
 from async_stagger import open_connection
 
 from .interface import ITCPTransport, ITCPReader, ITCPWriter
+from .params    import ClientTLSCertificate
 from .security  import tls_context
 
 class TCPReader(ITCPReader):
@@ -36,12 +37,13 @@ class TCPTransport(ITCPTransport):
             port:        int,
             tls:         bool,
             tls_verify:  bool=True,
+            certificate: Optional[ClientTLSCertificate]=None,
             bindhost:    Optional[str]=None
             ) -> Tuple[ITCPReader, ITCPWriter]:
 
         cur_ssl: Optional[SSLContext] = None
         if tls:
-            cur_ssl = tls_context(tls_verify)
+            cur_ssl = tls_context(tls_verify, certificate)
 
         local_addr: Optional[Tuple[str, int]] = None
         if not bindhost is None: