Use timezone-aware objects to represent datetimes in UTC

snitun/server/peer.py

@@ -1,6 +1,6 @@
 """Represent a single Peer."""
 import asyncio
-from datetime import datetime
+from datetime import datetime, timezone
 import hashlib
 import logging
 import os
@@ -60,7 +60,7 @@ def is_connected(self) -> bool:
     @property
     def is_valid(self) -> bool:
         """Return True if the peer is valid."""
-        return self._valid > datetime.utcnow()
+        return self._valid > datetime.now(tz=timezone.utc)
 
     @property
     def multiplexer(self) -> Optional[Multiplexer]:

snitun/server/peer_manager.py

@@ -1,6 +1,6 @@
 """Manage peer connections."""
 import asyncio
-from datetime import datetime
+from datetime import datetime, timezone
 import json
 import logging
 from typing import Callable, List, Optional
@@ -51,8 +51,8 @@ def create_peer(self, fernet_data: bytes) -> Peer:
             raise SniTunInvalidPeer("Invalid fernet token") from err
 
         # Check if token is valid
-        valid = datetime.utcfromtimestamp(config["valid"])
-        if valid < datetime.utcnow():
+        valid = datetime.fromtimestamp(config["valid"], tz=timezone.utc)
+        if valid < datetime.now(tz=timezone.utc):
             raise SniTunInvalidPeer("Token was expired")
 
         # Extract configuration

snitun/utils/server.py

@@ -1,5 +1,5 @@
 """Utils for server handling."""
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from typing import List
 import json
 
@@ -15,7 +15,7 @@ def generate_client_token(
 ):
     """Generate a token for client."""
     fernet = MultiFernet([Fernet(key) for key in tokens])
-    valid = datetime.utcnow() + valid_delta
+    valid = datetime.now(tz=timezone.utc) + valid_delta
 
     return fernet.encrypt(
         json.dumps(

tests/client/test_client_peer.py

@@ -1,6 +1,6 @@
 """Test Client Peer connections."""
 import asyncio
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 import ipaddress
 import os
 
@@ -23,7 +23,7 @@ async def test_init_client_peer(peer_listener, peer_manager, test_endpoint):
     assert not client.is_connected
     assert not peer_manager.peer_available("localhost")
 
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
     hostname = "localhost"
@@ -50,7 +50,7 @@ async def test_init_client_peer_with_alias(peer_listener, peer_manager, test_end
     assert not peer_manager.peer_available("localhost")
     assert not peer_manager.peer_available("localhost.custom")
 
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
     hostname = "localhost"
@@ -81,7 +81,7 @@ async def test_init_client_peer_invalid_token(
 
     assert not peer_manager.peer_available("localhost")
 
-    valid = datetime.utcnow() + timedelta(days=-1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=-1)
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
     hostname = "localhost"
@@ -100,7 +100,7 @@ async def test_flow_client_peer(peer_listener, peer_manager, test_endpoint):
 
     assert not peer_manager.peer_available("localhost")
 
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
     hostname = "localhost"
@@ -144,7 +144,7 @@ async def test_close_client_peer(peer_listener, peer_manager, test_endpoint):
 
     assert not peer_manager.peer_available("localhost")
 
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
     hostname = "localhost"
@@ -192,7 +192,7 @@ async def test_init_client_peer_wait(peer_listener, peer_manager, test_endpoint)
     assert not client.is_connected
     assert not peer_manager.peer_available("localhost")
 
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
     hostname = "localhost"
@@ -222,7 +222,7 @@ async def test_init_client_peer_throttling(peer_listener, peer_manager, test_end
     assert not client.is_connected
     assert not peer_manager.peer_available("localhost")
 
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
     hostname = "localhost"

tests/conftest.py

@@ -1,6 +1,6 @@
 """Pytest fixtures for SniTun."""
 import asyncio
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 import logging
 import os
 from unittest.mock import patch
@@ -221,7 +221,7 @@ def crypto_transport():
 @pytest.fixture
 async def peer(crypto_transport, multiplexer_server):
     """Init a peer with transport."""
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
     peer = Peer("localhost", valid, os.urandom(32), os.urandom(16))
     peer._crypto = crypto_transport
     peer._multiplexer = multiplexer_server

tests/server/test_all.py

@@ -1,6 +1,6 @@
 """Tests for peer listener & manager."""
 import asyncio
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 import hashlib
 import ipaddress
 import os
@@ -21,7 +21,7 @@ async def test_server_full(
     peer_messages = []
     peer_address = []
 
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
     hostname = "localhost"

tests/server/test_listener_peer.py

@@ -1,6 +1,6 @@
 """Tests for peer listener & manager."""
 import asyncio
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 import hashlib
 import os
 
@@ -24,7 +24,7 @@ async def test_init_listener(peer_manager):
 
 async def test_peer_listener(peer_manager, peer_listener, test_client_peer):
     """Run a full flow of with a peer."""
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
     hostname = "localhost"
@@ -47,7 +47,7 @@ async def test_peer_listener(peer_manager, peer_listener, test_client_peer):
 
 async def test_peer_listener_invalid(peer_manager, peer_listener, test_client_peer):
     """Run a full flow of with a peer."""
-    valid = datetime.utcnow() - timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) - timedelta(days=1)
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
     hostname = "localhost"
@@ -64,7 +64,7 @@ async def test_peer_listener_invalid(peer_manager, peer_listener, test_client_pe
 
 async def test_peer_listener_disconnect(peer_manager, peer_listener, test_client_peer):
     """Run a full flow of with a peer after that disconnect."""
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
     hostname = "localhost"
@@ -94,7 +94,7 @@ async def test_peer_listener_timeout(
     raise_timeout, peer_manager, peer_listener, test_client_peer
 ):
     """Run a full flow of with a peer."""
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
     hostname = "localhost"

tests/server/test_peer.py

@@ -1,5 +1,5 @@
 """Test a Peer object."""
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 import asyncio
 import hashlib
 import os
@@ -14,7 +14,7 @@
 
 def test_init_peer():
     """Test simple init of peer."""
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
     peer = Peer(
         "localhost",
         valid,
@@ -35,7 +35,7 @@ async def test_init_peer_multiplexer(event_loop, test_client, test_server):
     client = test_server[0]
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
 
     peer = Peer("localhost", valid, aes_key, aes_iv)
     crypto = CryptoTransport(aes_key, aes_iv)
@@ -77,7 +77,7 @@ async def test_init_peer_multiplexer_crypto(event_loop, test_client, test_server
     client = test_server[0]
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
 
     peer = Peer("localhost", valid, aes_key, aes_iv)
     crypto = CryptoTransport(aes_key, aes_iv)
@@ -129,7 +129,7 @@ async def test_init_peer_wrong_challenge(event_loop, test_client, test_server):
     client = test_server[0]
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
 
     peer = Peer("localhost", valid, aes_key, aes_iv)
     crypto = CryptoTransport(aes_key, aes_iv)
@@ -159,7 +159,7 @@ async def test_init_peer_wrong_challenge(event_loop, test_client, test_server):
 
 def test_init_peer_invalid():
     """Test simple init of peer with invalid date."""
-    valid = datetime.utcnow() - timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) - timedelta(days=1)
     peer = Peer("localhost", valid, os.urandom(32), os.urandom(16))
 
     assert not peer.is_valid
@@ -173,7 +173,7 @@ async def test_init_peer_multiplexer_throttling(event_loop, test_client, test_se
     client = test_server[0]
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
 
     peer = Peer("localhost", valid, aes_key, aes_iv, throttling=500)
     crypto = CryptoTransport(aes_key, aes_iv)

tests/server/test_peer_manager.py

@@ -1,7 +1,7 @@
 """Test peer manager."""
 import asyncio
 import os
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from unittest.mock import patch
 
 import pytest
@@ -26,7 +26,7 @@ async def test_init_new_peer():
     """Init a new peer."""
     manager = PeerManager(FERNET_TOKENS)
 
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
     hostname = "localhost"
@@ -51,7 +51,7 @@ async def test_init_new_peer_with_alias():
     """Init a new peer with custom domain."""
     manager = PeerManager(FERNET_TOKENS)
 
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
     hostname = "localhost"
@@ -86,7 +86,7 @@ async def test_init_new_peer_not_valid_time():
     """Init a new peer."""
     manager = PeerManager(FERNET_TOKENS)
 
-    valid = datetime.utcnow() - timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) - timedelta(days=1)
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
     hostname = "localhost"
@@ -108,7 +108,7 @@ async def test_init_new_peer_with_removing():
     """Init a new peer."""
     manager = PeerManager(FERNET_TOKENS)
 
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
     hostname = "localhost"
@@ -140,7 +140,7 @@ def _events(ev_peer: Peer, type_event: PeerManagerEvent) -> None:
 
     manager = PeerManager(FERNET_TOKENS, event_callback=_events)
 
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
     hostname = "localhost"
@@ -174,7 +174,7 @@ async def test_init_new_peer_throttling():
     """Init a new peer."""
     manager = PeerManager(FERNET_TOKENS, throttling=500)
 
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
     hostname = "localhost"
@@ -196,7 +196,7 @@ async def test_init_dual_peer_with_removing():
     """Init a new peer."""
     manager = PeerManager(FERNET_TOKENS)
 
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
     hostname = "localhost"
@@ -237,7 +237,7 @@ async def test_init_dual_peer_with_multiplexer(multiplexer_client):
     """Init a new peer."""
     manager = PeerManager(FERNET_TOKENS)
 
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
     hostname = "localhost"

tests/server/test_run.py

@@ -1,6 +1,6 @@
 """Test runner of SniTun Server."""
 import asyncio
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 import hashlib
 import ipaddress
 import os
@@ -69,7 +69,7 @@ async def test_snitun_single_runner():
         host="127.0.0.1", port="32000"
     )
 
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
     hostname = "localhost"
@@ -128,7 +128,7 @@ async def test_snitun_single_runner_timeout(raise_timeout):
         host="127.0.0.1", port="32000"
     )
 
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
     hostname = "localhost"
@@ -197,7 +197,7 @@ async def test_snitun_single_runner_throttling():
         host="127.0.0.1", port="32000"
     )
 
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
     hostname = "localhost"
@@ -264,7 +264,7 @@ def test_snitun_worker_runner(event_loop):
     sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
     sock.connect(("127.0.0.1", 32001))
 
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
     hostname = "localhost"
@@ -333,7 +333,7 @@ def test_snitun_worker_timeout(event_loop):
 
     time.sleep(1.5)
 
-    valid = datetime.utcnow() + timedelta(days=1)
+    valid = datetime.now(tz=timezone.utc) + timedelta(days=1)
     aes_key = os.urandom(32)
     aes_iv = os.urandom(16)
     hostname = "localhost"