We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Add token to websockets connector
Currently, anyone that knows opdroid url and endpoint will be able to request a socket to initialize a websocket connection.

Ideally, we should allow users to select a token in the configuration settings. When opsdroid gets a request, if the token doesn't exist then we just reject the request.

This will also a nice feature to implement along side opsdroid-web v2

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `opsdroid/connector/websocket/__init__.py`
Content:
```
1 """A connector which allows websocket connections."""
2 import json
3 import logging
4 import uuid
5 from datetime import datetime
6 
7 import aiohttp
8 import aiohttp.web
9 from aiohttp import WSCloseCode
10 from opsdroid.connector import Connector, register_event
11 from opsdroid.events import Message
12 
13 _LOGGER = logging.getLogger(__name__)
14 HEADERS = {"Access-Control-Allow-Origin": "*"}
15 CONFIG_SCHEMA = {"bot-name": str, "max-connections": int, "connection-timeout": int}
16 
17 
18 class ConnectorWebsocket(Connector):
19     """A connector which allows websocket connections."""
20 
21     def __init__(self, config, opsdroid=None):
22         """Create the connector."""
23         super().__init__(config, opsdroid=opsdroid)
24         _LOGGER.debug(_("Starting Websocket connector."))
25         self.name = config.get("name", "websocket")
26         self.max_connections = self.config.get("max-connections", 10)
27         self.connection_timeout = self.config.get("connection-timeout", 60)
28         self.accepting_connections = True
29         self.active_connections = {}
30         self.available_connections = []
31         self.bot_name = self.config.get("bot-name", "opsdroid")
32 
33     async def connect(self):
34         """Connect to the chat service."""
35         self.accepting_connections = True
36 
37         self.opsdroid.web_server.web_app.router.add_get(
38             "/connector/websocket/{socket}", self.websocket_handler
39         )
40 
41         self.opsdroid.web_server.web_app.router.add_post(
42             "/connector/websocket", self.new_websocket_handler
43         )
44 
45     async def disconnect(self):
46         """Disconnect from current sessions."""
47         self.accepting_connections = False
48         connections_to_close = self.active_connections.copy()
49         for connection in connections_to_close:
50             await connections_to_close[connection].close(
51                 code=WSCloseCode.GOING_AWAY, message="Server shutdown"
52             )
53 
54     async def new_websocket_handler(self, request):
55         """Handle for aiohttp creating websocket connections."""
56         if (
57             len(self.active_connections) + len(self.available_connections)
58             < self.max_connections
59             and self.accepting_connections
60         ):
61             socket = {"id": str(uuid.uuid1()), "date": datetime.now()}
62             self.available_connections.append(socket)
63             return aiohttp.web.Response(
64                 text=json.dumps({"socket": socket["id"]}), headers=HEADERS, status=200
65             )
66         return aiohttp.web.Response(
67             text=json.dumps("No connections available"), headers=HEADERS, status=429
68         )
69 
70     async def websocket_handler(self, request):
71         """Handle for aiohttp handling websocket connections."""
72         socket = request.match_info.get("socket")
73         available = [
74             item for item in self.available_connections if item["id"] == socket
75         ]
76         if len(available) != 1:
77             return aiohttp.web.Response(
78                 text=json.dumps("Please request a socket first"),
79                 headers=HEADERS,
80                 status=400,
81             )
82         if (
83             datetime.now() - available[0]["date"]
84         ).total_seconds() > self.connection_timeout:
85             self.available_connections.remove(available[0])
86             return aiohttp.web.Response(
87                 text=json.dumps("Socket request timed out"), headers=HEADERS, status=408
88             )
89         self.available_connections.remove(available[0])
90         _LOGGER.debug(_("User connected to %s."), socket)
91 
92         websocket = aiohttp.web.WebSocketResponse()
93         await websocket.prepare(request)
94 
95         self.active_connections[socket] = websocket
96         async for msg in websocket:
97             if msg.type == aiohttp.WSMsgType.TEXT:
98                 message = Message(text=msg.data, user=None, target=None, connector=self)
99                 await self.opsdroid.parse(message)
100             elif msg.type == aiohttp.WSMsgType.ERROR:
101                 _LOGGER.error(
102                     _("Websocket connection closed with exception %s."),
103                     websocket.exception(),
104                 )
105 
106         _LOGGER.info(_("websocket connection closed"))
107         self.active_connections.pop(socket, None)
108 
109         return websocket
110 
111     async def listen(self):
112         """Listen for and parse new messages.
113 
114         Listening is handled by the aiohttp web server so
115         we don't need to do anything here.
116 
117         """
118 
119     @register_event(Message)
120     async def send_message(self, message):
121         """Respond with a message."""
122         try:
123             if message.target is None:
124                 message.target = next(iter(self.active_connections))
125             _LOGGER.debug(
126                 _("Responding with: '%s' in target %s"), message.text, message.target
127             )
128             await self.active_connections[message.target].send_str(message.text)
129         except KeyError:
130             _LOGGER.error(_("No active socket for target %s"), message.target)
131
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/opsdroid/connector/websocket/__init__.py b/opsdroid/connector/websocket/__init__.py
--- a/opsdroid/connector/websocket/__init__.py
+++ b/opsdroid/connector/websocket/__init__.py
@@ -9,12 +9,49 @@
 from aiohttp import WSCloseCode
 from opsdroid.connector import Connector, register_event
 from opsdroid.events import Message
+import dataclasses
+from typing import Optional
 
 _LOGGER = logging.getLogger(__name__)
 HEADERS = {"Access-Control-Allow-Origin": "*"}
 CONFIG_SCHEMA = {"bot-name": str, "max-connections": int, "connection-timeout": int}
 
 
+@dataclasses.dataclass
+class WebsocketMessage:
+    """A message received from a websocket connection."""
+
+    message: str
+    user: Optional[str]
+    socket: Optional[str]
+
+    @classmethod
+    def parse_payload(cls, payload: str):
+        """Parse the payload of a websocket message.
+
+        We will try to parse the payload as a json string.
+        If that fails, we will use the default values which are:
+
+        message: str
+        user: None
+        socket: None
+
+        """
+        try:
+            data = json.loads(payload)
+            return cls(
+                message=data.get("message"),
+                user=data.get("user"),
+                socket=data.get("socket"),
+            )
+        except json.JSONDecodeError:
+            return cls(
+                message=payload,
+                user=None,
+                socket=None,
+            )
+
+
 class ConnectorWebsocket(Connector):
     """A connector which allows websocket connections."""
 
@@ -29,6 +66,7 @@
         self.active_connections = {}
         self.available_connections = []
         self.bot_name = self.config.get("bot-name", "opsdroid")
+        self.authorization_token = self.config.get("token")
 
     async def connect(self):
         """Connect to the chat service."""
@@ -53,6 +91,7 @@
 
     async def new_websocket_handler(self, request):
         """Handle for aiohttp creating websocket connections."""
+        await self.validate_request(request)
         if (
             len(self.active_connections) + len(self.available_connections)
             < self.max_connections
@@ -95,7 +134,13 @@
         self.active_connections[socket] = websocket
         async for msg in websocket:
             if msg.type == aiohttp.WSMsgType.TEXT:
-                message = Message(text=msg.data, user=None, target=None, connector=self)
+                payload = WebsocketMessage.parse_payload(msg.data)
+                message = Message(
+                    text=payload.message,
+                    user=payload.user,
+                    target=payload.socket,
+                    connector=self,
+                )
                 await self.opsdroid.parse(message)
             elif msg.type == aiohttp.WSMsgType.ERROR:
                 _LOGGER.error(
@@ -108,6 +153,20 @@
 
         return websocket
 
+    async def validate_request(self, request):
+        """Validate the request by looking at headers and the connector token.
+
+        If the token does not exist in the header, but exists in the configuration,
+        then we will simply return a Forbidden error.
+
+        """
+        client_token = request.headers.get("Authorization")
+        if self.authorization_token and (
+            client_token is None or client_token != self.authorization_token
+        ):
+            raise aiohttp.web.HTTPUnauthorized()
+        return True
+
     async def listen(self):
         """Listen for and parse new messages.
 
@@ -117,7 +176,7 @@
         """
 
     @register_event(Message)
-    async def send_message(self, message):
+    async def send_message(self, message: Message):
         """Respond with a message."""
         try:
             if message.target is None:

{"golden_diff": "diff --git a/opsdroid/connector/websocket/__init__.py b/opsdroid/connector/websocket/__init__.py\n--- a/opsdroid/connector/websocket/__init__.py\n+++ b/opsdroid/connector/websocket/__init__.py\n@@ -9,12 +9,49 @@\n from aiohttp import WSCloseCode\n from opsdroid.connector import Connector, register_event\n from opsdroid.events import Message\n+import dataclasses\n+from typing import Optional\n \n _LOGGER = logging.getLogger(__name__)\n HEADERS = {\"Access-Control-Allow-Origin\": \"*\"}\n CONFIG_SCHEMA = {\"bot-name\": str, \"max-connections\": int, \"connection-timeout\": int}\n \n \n+@dataclasses.dataclass\n+class WebsocketMessage:\n+    \"\"\"A message received from a websocket connection.\"\"\"\n+\n+    message: str\n+    user: Optional[str]\n+    socket: Optional[str]\n+\n+    @classmethod\n+    def parse_payload(cls, payload: str):\n+        \"\"\"Parse the payload of a websocket message.\n+\n+        We will try to parse the payload as a json string.\n+        If that fails, we will use the default values which are:\n+\n+        message: str\n+        user: None\n+        socket: None\n+\n+        \"\"\"\n+        try:\n+            data = json.loads(payload)\n+            return cls(\n+                message=data.get(\"message\"),\n+                user=data.get(\"user\"),\n+                socket=data.get(\"socket\"),\n+            )\n+        except json.JSONDecodeError:\n+            return cls(\n+                message=payload,\n+                user=None,\n+                socket=None,\n+            )\n+\n+\n class ConnectorWebsocket(Connector):\n     \"\"\"A connector which allows websocket connections.\"\"\"\n \n@@ -29,6 +66,7 @@\n         self.active_connections = {}\n         self.available_connections = []\n         self.bot_name = self.config.get(\"bot-name\", \"opsdroid\")\n+        self.authorization_token = self.config.get(\"token\")\n \n     async def connect(self):\n         \"\"\"Connect to the chat service.\"\"\"\n@@ -53,6 +91,7 @@\n \n     async def new_websocket_handler(self, request):\n         \"\"\"Handle for aiohttp creating websocket connections.\"\"\"\n+        await self.validate_request(request)\n         if (\n             len(self.active_connections) + len(self.available_connections)\n             < self.max_connections\n@@ -95,7 +134,13 @@\n         self.active_connections[socket] = websocket\n         async for msg in websocket:\n             if msg.type == aiohttp.WSMsgType.TEXT:\n-                message = Message(text=msg.data, user=None, target=None, connector=self)\n+                payload = WebsocketMessage.parse_payload(msg.data)\n+                message = Message(\n+                    text=payload.message,\n+                    user=payload.user,\n+                    target=payload.socket,\n+                    connector=self,\n+                )\n                 await self.opsdroid.parse(message)\n             elif msg.type == aiohttp.WSMsgType.ERROR:\n                 _LOGGER.error(\n@@ -108,6 +153,20 @@\n \n         return websocket\n \n+    async def validate_request(self, request):\n+        \"\"\"Validate the request by looking at headers and the connector token.\n+\n+        If the token does not exist in the header, but exists in the configuration,\n+        then we will simply return a Forbidden error.\n+\n+        \"\"\"\n+        client_token = request.headers.get(\"Authorization\")\n+        if self.authorization_token and (\n+            client_token is None or client_token != self.authorization_token\n+        ):\n+            raise aiohttp.web.HTTPUnauthorized()\n+        return True\n+\n     async def listen(self):\n         \"\"\"Listen for and parse new messages.\n \n@@ -117,7 +176,7 @@\n         \"\"\"\n \n     @register_event(Message)\n-    async def send_message(self, message):\n+    async def send_message(self, message: Message):\n         \"\"\"Respond with a message.\"\"\"\n         try:\n             if message.target is None:\n", "issue": "Add token to websockets connector\nCurrently, anyone that knows opdroid url and endpoint will be able to request a socket to initialize a websocket connection.\r\n\r\nIdeally, we should allow users to select a token in the configuration settings. When opsdroid gets a request, if the token doesn't exist then we just reject the request.\r\n\r\nThis will also a nice feature to implement along side opsdroid-web v2\n", "before_files": [{"content": "\"\"\"A connector which allows websocket connections.\"\"\"\nimport json\nimport logging\nimport uuid\nfrom datetime import datetime\n\nimport aiohttp\nimport aiohttp.web\nfrom aiohttp import WSCloseCode\nfrom opsdroid.connector import Connector, register_event\nfrom opsdroid.events import Message\n\n_LOGGER = logging.getLogger(__name__)\nHEADERS = {\"Access-Control-Allow-Origin\": \"*\"}\nCONFIG_SCHEMA = {\"bot-name\": str, \"max-connections\": int, \"connection-timeout\": int}\n\n\nclass ConnectorWebsocket(Connector):\n    \"\"\"A connector which allows websocket connections.\"\"\"\n\n    def __init__(self, config, opsdroid=None):\n        \"\"\"Create the connector.\"\"\"\n        super().__init__(config, opsdroid=opsdroid)\n        _LOGGER.debug(_(\"Starting Websocket connector.\"))\n        self.name = config.get(\"name\", \"websocket\")\n        self.max_connections = self.config.get(\"max-connections\", 10)\n        self.connection_timeout = self.config.get(\"connection-timeout\", 60)\n        self.accepting_connections = True\n        self.active_connections = {}\n        self.available_connections = []\n        self.bot_name = self.config.get(\"bot-name\", \"opsdroid\")\n\n    async def connect(self):\n        \"\"\"Connect to the chat service.\"\"\"\n        self.accepting_connections = True\n\n        self.opsdroid.web_server.web_app.router.add_get(\n            \"/connector/websocket/{socket}\", self.websocket_handler\n        )\n\n        self.opsdroid.web_server.web_app.router.add_post(\n            \"/connector/websocket\", self.new_websocket_handler\n        )\n\n    async def disconnect(self):\n        \"\"\"Disconnect from current sessions.\"\"\"\n        self.accepting_connections = False\n        connections_to_close = self.active_connections.copy()\n        for connection in connections_to_close:\n            await connections_to_close[connection].close(\n                code=WSCloseCode.GOING_AWAY, message=\"Server shutdown\"\n            )\n\n    async def new_websocket_handler(self, request):\n        \"\"\"Handle for aiohttp creating websocket connections.\"\"\"\n        if (\n            len(self.active_connections) + len(self.available_connections)\n            < self.max_connections\n            and self.accepting_connections\n        ):\n            socket = {\"id\": str(uuid.uuid1()), \"date\": datetime.now()}\n            self.available_connections.append(socket)\n            return aiohttp.web.Response(\n                text=json.dumps({\"socket\": socket[\"id\"]}), headers=HEADERS, status=200\n            )\n        return aiohttp.web.Response(\n            text=json.dumps(\"No connections available\"), headers=HEADERS, status=429\n        )\n\n    async def websocket_handler(self, request):\n        \"\"\"Handle for aiohttp handling websocket connections.\"\"\"\n        socket = request.match_info.get(\"socket\")\n        available = [\n            item for item in self.available_connections if item[\"id\"] == socket\n        ]\n        if len(available) != 1:\n            return aiohttp.web.Response(\n                text=json.dumps(\"Please request a socket first\"),\n                headers=HEADERS,\n                status=400,\n            )\n        if (\n            datetime.now() - available[0][\"date\"]\n        ).total_seconds() > self.connection_timeout:\n            self.available_connections.remove(available[0])\n            return aiohttp.web.Response(\n                text=json.dumps(\"Socket request timed out\"), headers=HEADERS, status=408\n            )\n        self.available_connections.remove(available[0])\n        _LOGGER.debug(_(\"User connected to %s.\"), socket)\n\n        websocket = aiohttp.web.WebSocketResponse()\n        await websocket.prepare(request)\n\n        self.active_connections[socket] = websocket\n        async for msg in websocket:\n            if msg.type == aiohttp.WSMsgType.TEXT:\n                message = Message(text=msg.data, user=None, target=None, connector=self)\n                await self.opsdroid.parse(message)\n            elif msg.type == aiohttp.WSMsgType.ERROR:\n                _LOGGER.error(\n                    _(\"Websocket connection closed with exception %s.\"),\n                    websocket.exception(),\n                )\n\n        _LOGGER.info(_(\"websocket connection closed\"))\n        self.active_connections.pop(socket, None)\n\n        return websocket\n\n    async def listen(self):\n        \"\"\"Listen for and parse new messages.\n\n        Listening is handled by the aiohttp web server so\n        we don't need to do anything here.\n\n        \"\"\"\n\n    @register_event(Message)\n    async def send_message(self, message):\n        \"\"\"Respond with a message.\"\"\"\n        try:\n            if message.target is None:\n                message.target = next(iter(self.active_connections))\n            _LOGGER.debug(\n                _(\"Responding with: '%s' in target %s\"), message.text, message.target\n            )\n            await self.active_connections[message.target].send_str(message.text)\n        except KeyError:\n            _LOGGER.error(_(\"No active socket for target %s\"), message.target)\n", "path": "opsdroid/connector/websocket/__init__.py"}], "after_files": [{"content": "\"\"\"A connector which allows websocket connections.\"\"\"\nimport json\nimport logging\nimport uuid\nfrom datetime import datetime\n\nimport aiohttp\nimport aiohttp.web\nfrom aiohttp import WSCloseCode\nfrom opsdroid.connector import Connector, register_event\nfrom opsdroid.events import Message\nimport dataclasses\nfrom typing import Optional\n\n_LOGGER = logging.getLogger(__name__)\nHEADERS = {\"Access-Control-Allow-Origin\": \"*\"}\nCONFIG_SCHEMA = {\"bot-name\": str, \"max-connections\": int, \"connection-timeout\": int}\n\n\n@dataclasses.dataclass\nclass WebsocketMessage:\n    \"\"\"A message received from a websocket connection.\"\"\"\n\n    message: str\n    user: Optional[str]\n    socket: Optional[str]\n\n    @classmethod\n    def parse_payload(cls, payload: str):\n        \"\"\"Parse the payload of a websocket message.\n\n        We will try to parse the payload as a json string.\n        If that fails, we will use the default values which are:\n\n        message: str\n        user: None\n        socket: None\n\n        \"\"\"\n        try:\n            data = json.loads(payload)\n            return cls(\n                message=data.get(\"message\"),\n                user=data.get(\"user\"),\n                socket=data.get(\"socket\"),\n            )\n        except json.JSONDecodeError:\n            return cls(\n                message=payload,\n                user=None,\n                socket=None,\n            )\n\n\nclass ConnectorWebsocket(Connector):\n    \"\"\"A connector which allows websocket connections.\"\"\"\n\n    def __init__(self, config, opsdroid=None):\n        \"\"\"Create the connector.\"\"\"\n        super().__init__(config, opsdroid=opsdroid)\n        _LOGGER.debug(_(\"Starting Websocket connector.\"))\n        self.name = config.get(\"name\", \"websocket\")\n        self.max_connections = self.config.get(\"max-connections\", 10)\n        self.connection_timeout = self.config.get(\"connection-timeout\", 60)\n        self.accepting_connections = True\n        self.active_connections = {}\n        self.available_connections = []\n        self.bot_name = self.config.get(\"bot-name\", \"opsdroid\")\n        self.authorization_token = self.config.get(\"token\")\n\n    async def connect(self):\n        \"\"\"Connect to the chat service.\"\"\"\n        self.accepting_connections = True\n\n        self.opsdroid.web_server.web_app.router.add_get(\n            \"/connector/websocket/{socket}\", self.websocket_handler\n        )\n\n        self.opsdroid.web_server.web_app.router.add_post(\n            \"/connector/websocket\", self.new_websocket_handler\n        )\n\n    async def disconnect(self):\n        \"\"\"Disconnect from current sessions.\"\"\"\n        self.accepting_connections = False\n        connections_to_close = self.active_connections.copy()\n        for connection in connections_to_close:\n            await connections_to_close[connection].close(\n                code=WSCloseCode.GOING_AWAY, message=\"Server shutdown\"\n            )\n\n    async def new_websocket_handler(self, request):\n        \"\"\"Handle for aiohttp creating websocket connections.\"\"\"\n        await self.validate_request(request)\n        if (\n            len(self.active_connections) + len(self.available_connections)\n            < self.max_connections\n            and self.accepting_connections\n        ):\n            socket = {\"id\": str(uuid.uuid1()), \"date\": datetime.now()}\n            self.available_connections.append(socket)\n            return aiohttp.web.Response(\n                text=json.dumps({\"socket\": socket[\"id\"]}), headers=HEADERS, status=200\n            )\n        return aiohttp.web.Response(\n            text=json.dumps(\"No connections available\"), headers=HEADERS, status=429\n        )\n\n    async def websocket_handler(self, request):\n        \"\"\"Handle for aiohttp handling websocket connections.\"\"\"\n        socket = request.match_info.get(\"socket\")\n        available = [\n            item for item in self.available_connections if item[\"id\"] == socket\n        ]\n        if len(available) != 1:\n            return aiohttp.web.Response(\n                text=json.dumps(\"Please request a socket first\"),\n                headers=HEADERS,\n                status=400,\n            )\n        if (\n            datetime.now() - available[0][\"date\"]\n        ).total_seconds() > self.connection_timeout:\n            self.available_connections.remove(available[0])\n            return aiohttp.web.Response(\n                text=json.dumps(\"Socket request timed out\"), headers=HEADERS, status=408\n            )\n        self.available_connections.remove(available[0])\n        _LOGGER.debug(_(\"User connected to %s.\"), socket)\n\n        websocket = aiohttp.web.WebSocketResponse()\n        await websocket.prepare(request)\n\n        self.active_connections[socket] = websocket\n        async for msg in websocket:\n            if msg.type == aiohttp.WSMsgType.TEXT:\n                payload = WebsocketMessage.parse_payload(msg.data)\n                message = Message(\n                    text=payload.message,\n                    user=payload.user,\n                    target=payload.socket,\n                    connector=self,\n                )\n                await self.opsdroid.parse(message)\n            elif msg.type == aiohttp.WSMsgType.ERROR:\n                _LOGGER.error(\n                    _(\"Websocket connection closed with exception %s.\"),\n                    websocket.exception(),\n                )\n\n        _LOGGER.info(_(\"websocket connection closed\"))\n        self.active_connections.pop(socket, None)\n\n        return websocket\n\n    async def validate_request(self, request):\n        \"\"\"Validate the request by looking at headers and the connector token.\n\n        If the token does not exist in the header, but exists in the configuration,\n        then we will simply return a Forbidden error.\n\n        \"\"\"\n        client_token = request.headers.get(\"Authorization\")\n        if self.authorization_token and (\n            client_token is None or client_token != self.authorization_token\n        ):\n            raise aiohttp.web.HTTPUnauthorized()\n        return True\n\n    async def listen(self):\n        \"\"\"Listen for and parse new messages.\n\n        Listening is handled by the aiohttp web server so\n        we don't need to do anything here.\n\n        \"\"\"\n\n    @register_event(Message)\n    async def send_message(self, message: Message):\n        \"\"\"Respond with a message.\"\"\"\n        try:\n            if message.target is None:\n                message.target = next(iter(self.active_connections))\n            _LOGGER.debug(\n                _(\"Responding with: '%s' in target %s\"), message.text, message.target\n            )\n            await self.active_connections[message.target].send_str(message.text)\n        except KeyError:\n            _LOGGER.error(_(\"No active socket for target %s\"), message.target)\n", "path": "opsdroid/connector/websocket/__init__.py"}]}