mirror
/
ValvePython.steam
mirror of https://github.com/ValvePython/steam
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

add utils.ip6_*; code changes due to proto update

pull/293/head
Rossen Georgiev 5 years ago
parent
e584dff5e3
commit
f5690e6b22
6 changed files with 67 additions and 33 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 6
      steam/client/__init__.py
  2. 2
      steam/client/builtins/apps.py
  3. 15
      steam/client/builtins/gameservers.py
  4. 4
      steam/core/cm.py
  5. 40
      steam/utils/__init__.py
  6. 33
      tests/test_utils.py

6
steam/client/__init__.py
View File

@ -31,7 +31,7 @@ from steam.core.crypto import sha1_hash
from steam.steamid import SteamID
from steam.exceptions import SteamError
from steam.client.builtins import BuiltinBase
from steam.utils import ip_from_int, ip_to_int
from steam.utils import ip4_from_int, ip4_to_int
from steam.utils.proto import proto_fill_from_dict
if six.PY2:
@ -174,7 +174,7 @@ class SteamClient(CMClient, BuiltinBase):
data = {
'cell_id': self.cm_servers.cell_id,
'last_updated': self.cm_servers.last_updated,
'servers': list(zip(map(ip_from_int, msg.body.cm_addresses), msg.body.cm_ports)),
'servers': list(zip(map(ip4_from_int, msg.body.cm_addresses), msg.body.cm_ports)),
}
try:
with open(filepath, 'wb') as f:
@ -539,7 +539,7 @@ class SteamClient(CMClient, BuiltinBase):
message.body.chat_mode = self.chat_mode
if login_id is None:
message.body.obfuscated_private_ip.v4 = ip_to_int(self.connection.local_address) ^ 0xF00DBAAD
message.body.obfuscated_private_ip.v4 = ip4_to_int(self.connection.local_address) ^ 0xF00DBAAD
else:
message.body.obfuscated_private_ip.v4 = login_id

2
steam/client/builtins/apps.py
View File

@ -3,7 +3,6 @@ import vdf
from steam.enums import EResult, EServerType
from steam.enums.emsg import EMsg
from steam.core.msg import MsgProto
from steam.utils import ip_from_int
from steam.utils.proto import proto_fill_from_dict
@ -108,7 +107,6 @@ class Apps(object):
for app in apps:
app_info = message.body.apps.add()
app_info.only_public = False
if tokens:
app_info.access_token = tokens['apps'].get(app['appid'] if isinstance(app, dict) else app, 0)

15
steam/client/builtins/gameservers.py
View File

@ -38,7 +38,7 @@ from steam.steamid import SteamID
from steam.core.msg import MsgProto
from steam.enums import EResult
from steam.enums.emsg import EMsg
from steam.utils import ip_to_int, ip_from_int
from steam.utils import ip4_to_int, ip4_from_int, ip6_from_bytes
from steam.utils.proto import proto_to_dict
from steam.exceptions import SteamError
@ -81,13 +81,13 @@ class SteamGameServers(object):
.. code:: python
[{'auth_players': 0, 'server_ip': '1.2.3.4', 'server_port': 27015},
{'auth_players': 6, 'server_ip': '1.2.3.4', 'server_port': 27016},
[{'auth_players': 0, 'server_ip': '1.2.3.4', 'query_port': 27015},
{'auth_players': 6, 'server_ip': '1:2:3:4::5', 'query_port': 27016},
...
]
"""
if 'geo_location_ip' in kw:
kw['geo_location_ip'] = ip_to_int(kw['geo_location_ip'])
kw['geo_location_ip'] = ip4_to_int(kw['geo_location_ip'])
kw['filter_text'] = filter_text
kw['max_servers'] = max_servers
@ -103,7 +103,12 @@ class SteamGameServers(object):
resp = proto_to_dict(resp)
for server in resp['servers']:
server['server_ip'] = ip_from_int(server['server_ip'])
server.pop('deprecated_server_ip', None) # no point returning this
if 'v4' in server['server_ip']:
server['server_ip'] = ip4_from_int(server['server_ip']['v4'])
else:
server['server_ip'] = ip6_from_bytes(server['server_ip']['v6'])
return resp['servers']

4
steam/core/cm.py
View File

@ -19,7 +19,7 @@ from steam.core import crypto
from steam.core.connection import TCPConnection
from steam.core.msg import Msg, MsgProto
from eventemitter import EventEmitter
from steam.utils import ip_from_int
from steam.utils import ip4_from_int
from steam.utils.proto import is_proto, clear_proto_bit
@ -392,7 +392,7 @@ class CMClient(EventEmitter):
def _handle_cm_list(self, msg):
self._LOG.debug("Updating CM list")
new_servers = zip(map(ip_from_int, msg.body.cm_addresses), msg.body.cm_ports)
new_servers = zip(map(ip4_from_int, msg.body.cm_addresses), msg.body.cm_ports)
self.cm_servers.clear()
self.cm_servers.merge_list(new_servers)
self.cm_servers.cell_id = self.cell_id

40
steam/utils/__init__.py
View File

@ -7,25 +7,45 @@ import sys
from six.moves import xrange as _range
def ip_from_int(ip):
"""Convert IP to :py:class:`int`
def ip4_from_int(ip):
"""Convert :py:class:`int` to IPv4 string
:param ip: IP in dot-decimal notation
:param ip: int representing an IPv4
:type ip: int
:return: IP in dot-decimal notation
:rtype: str
"""
return socket.inet_ntoa(struct.pack(">L", ip))
def ip4_to_int(ip):
"""Convert IPv4 string to :py:class:`int`
:param ip: IPv4 in dot-decimal notation
:type ip: str
:rtype: int
"""
return socket.inet_ntoa(struct.pack(">L", ip))
return struct.unpack(">L", socket.inet_aton(ip))[0]
def ip_to_int(ip):
"""Convert :py:class:`int` to IP
ip_to_int = ip4_to_int
ip_from_int = ip4_from_int
:param ip: int representing an IP
:type ip: int
:return: IP in dot-decimal notation
def ip6_from_bytes(ip):
"""Convert :py:class:`bytes` to IPv6 string
:param ip: IPv6 in dot-decimal notation
:type ip: bytes
:rtype: str
"""
return struct.unpack(">L", socket.inet_aton(ip))[0]
return socket.inet_ntop(socket.AF_INET6, ip)
def ip6_to_bytes(ip):
"""Convert IPv6 string to :py:class:`bytes`
:param ip: IPv6 in dot-decimal notation
:type ip: str
:rtype: bytes
"""
return socket.inet_pton(socket.AF_INET6, ip)
def chunks(arr, size):

33
tests/test_utils.py
View File

@ -5,22 +5,33 @@ import steam.utils.web as uweb
import requests
from steam.protobufs.test_messages_pb2 import ComplexProtoMessage
proto_mask = 0x80000000
class Util_Functions(unittest.TestCase):
def test_ip_from_int(self):
self.assertEqual('0.0.0.0', ut.ip_from_int(0))
self.assertEqual('12.34.56.78', ut.ip_from_int(203569230))
self.assertEqual('255.255.255.255', ut.ip_from_int(4294967295))
def test_ip_to_int(self):
self.assertEqual(ut.ip_to_int('0.0.0.0'), 0)
self.assertEqual(ut.ip_to_int('12.34.56.78'), 203569230)
self.assertEqual(ut.ip_to_int('255.255.255.255'), 4294967295)
def test_ip4_from_int(self):
self.assertEqual('0.0.0.0', ut.ip4_from_int(0))
self.assertEqual('12.34.56.78', ut.ip4_from_int(203569230))
self.assertEqual('255.255.255.255', ut.ip4_from_int(4294967295))
def test_ip4_to_int(self):
self.assertEqual(ut.ip4_to_int('0.0.0.0'), 0)
self.assertEqual(ut.ip4_to_int('12.34.56.78'), 203569230)
self.assertEqual(ut.ip4_to_int('255.255.255.255'), 4294967295)
def test_ip6_from_bytes(self):
self.assertEqual('::1', ut.ip6_from_bytes(b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01'))
self.assertEqual('1:2:3:4:5:6:7:8', ut.ip6_from_bytes(b'\x00\x01\x00\x02\x00\x03\x00\x04\x00\x05\x00\x06\x00\x07\x00\x08'))
self.assertEqual('1234:5678:9abc:def0:0:dead:beef:1', ut.ip6_from_bytes(b'\x12\x34\x56\x78\x9a\xbc\xde\xf0\x00\x00\xde\xad\xbe\xef\x00\x01'))
def test_ip6_to_bytes(self):
self.assertEqual(ut.ip6_to_bytes('::1'), b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01')
self.assertEqual(ut.ip6_to_bytes('1:2:3:4:5:6:7:8'), b'\x00\x01\x00\x02\x00\x03\x00\x04\x00\x05\x00\x06\x00\x07\x00\x08')
self.assertEqual(ut.ip6_to_bytes('1234:5678:9abc:def0:0:dead:beef:1'), b'\x12\x34\x56\x78\x9a\xbc\xde\xf0\x00\x00\xde\xad\xbe\xef\x00\x01')
def test_make_requests_session(self):
self.assertIsInstance(uweb.make_requests_session(), requests.Session)
proto_mask = 0x80000000
class Util_Proto_Functions(unittest.TestCase):
def test_is_proto(self):
self.assertTrue(utp.is_proto(proto_mask))

Write Preview
Loading…
Cancel
Save