python-ipware/python_ipware/python_ipware.py at main · un33k/python-ipware · GitHub

History

335 lines (277 loc) · 12.4 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

import ipaddress

import logging

from typing import Dict, List, Optional, Tuple, Union

IpAddressType = Union[ipaddress.IPv4Address, ipaddress.IPv6Address]

OptionalIpAddressType = Optional[IpAddressType]

logger = logging.getLogger(__name__)

class IpWareMeta:

"""

A class to manage metadata extracted from HTTP request headers, primarily for identifying

the client's IP address behind load balancers and proxies.

"""

def __init__(

self, precedence: Optional[Tuple[str, ...]] = None, leftmost: bool = True

) -> None:

"""

Initializes the metadata handler with a precedence list of HTTP headers and the position of the client IP.

:param precedence: A tuple of header names that define the order to check for a valid IP.

If not provided, a default list of commonly used headers is applied.

:param leftmost: A boolean that indicates whether to use the left-most IP in the header

when multiple IPs are listed (typically the case with proxies).

"""

# Default precedence list of headers if none is provided

if precedence is None:

precedence = (

"X_FORWARDED_FOR", # Common header for proxies, load balancers, like AWS ELB. Default to the left-most IP.

"HTTP_X_FORWARDED_FOR", # Alternative header similar to `X_FORWARDED_FOR`.

"HTTP_CLIENT_IP", # Header used by some providers like Amazon EC2, Heroku.

"HTTP_X_REAL_IP", # Header used by some providers like Amazon EC2, Heroku.

"HTTP_X_FORWARDED", # Used by Squid and similar software.

"HTTP_X_CLUSTER_CLIENT_IP", # Used by Rackspace LB, Riverbed Stingray.

"HTTP_FORWARDED_FOR", # Standard header defined by RFC 7239.

"HTTP_FORWARDED", # Standard header defined by RFC 7239.

"HTTP_CF_CONNECTING_IP", # Used by CloudFlare.

"X-CLIENT-IP", # Used by Microsoft Azure.

"X-REAL-IP", # Commonly used by NGINX.

"X-CLUSTER-CLIENT-IP", # Used by Rackspace Cloud Load Balancers.

"X_FORWARDED", # Used by Squid.

"FORWARDED_FOR", # Standard header defined by RFC 7239.

"CF-CONNECTING-IP", # Used by CloudFlare.

"TRUE-CLIENT-IP", # Header for CloudFlare Enterprise.

"FASTLY-CLIENT-IP", # Used by Fastly, Firebase.

"FORWARDED", # Standard header defined by RFC 7239.

"CLIENT-IP", # Used by Akamai, Cloudflare's True-Client-IP, and Fastly's Fastly-Client-IP.

"REMOTE_ADDR", # The default IP address header (direct connection).

)

self.precedence = precedence

self.leftmost = leftmost

class IpWareIpAddress:

"""

A class for handling and parsing IP address data from HTTP requests.

"""

def extract_ipv4(self, ip_address: str) -> str:

"""

Extracts the IPv4 address from a given string that may include a port number.

Args:

ip_address (str): An IPv4 address possibly including a port (e.g., "192.168.1.1:8080").

Returns:

str: The IPv4 address without the port. Returns an empty string if input is None.

"""

if ip_address:

return ip_address.split(":")[0].strip()

return ""

def extract_ipv6(self, ip_address: str) -> str:

"""

Extracts the IPv6 address from a given string that may include a port number.

Args:

ip_address (str): An IPv6 address possibly including a port (e.g., "[2001:db8::1]:8080").

Returns:

str: The IPv6 address without brackets or port. Returns an empty string if input is None.

"""

if ip_address and "]:" in ip_address:

return ip_address.split("]:")[0].replace("[", "").strip()

return ip_address.strip() if ip_address else ""

def parse_ip_address(self, ip_str: str):

"""

Parses the given IP address string to an IPv4Address or IPv6Address object.

Args:

ip_str (str): An IP address possibly including a port.

Returns:

ipaddress.IPv4Address|ipaddress.IPv6Address|None: The parsed IP object or None if the address is invalid.

"""

try:

# First, try to parse as IPv6.

ipv6 = self.extract_ipv6(ip_str)

ip = ipaddress.IPv6Address(ipv6)

return ip.ipv4_mapped or ip

except ipaddress.AddressValueError:

try:

# If IPv6 parsing fails, try to parse as IPv4.

ipv4 = self.extract_ipv4(ip_str)

return ipaddress.IPv4Address(ipv4)

except ipaddress.AddressValueError:

# Log error if IP is invalid

# print(f"Invalid IP address: {ip_str}")

return None

def get_ips_from_string(

self,

ip_str: str,

strict: bool = False,

) -> Optional[List[Union[ipaddress.IPv4Address, ipaddress.IPv6Address]]]:

"""

Parses a comma-separated list of IP addresses, each possibly including a port.

Args:

ip_str (str): A comma-separated list of IP addresses or address:port entries.

strict (bool): True to bail out on first invalid ip, False to skip invalid ip

Returns:

List[ipaddress.IPv4Address|ipaddress.IPv6Address]|None: A list of IP address objects or None if any IP is invalid.

"""

ip_list = []

for ip_address in ip_str.split(","):

ip = self.parse_ip_address(ip_address.strip())

if ip:

ip_list.append(ip)

else:

if strict:

return None

if not self.leftmost:

ip_list.reverse()

return ip_list

class IpWareProxy:

"""

A class to handle proxy data from an HTTP request, including validating

proxy counts and trusted proxy lists.

"""

def __init__(

self, proxy_count: Optional[int] = None, proxy_list: Optional[List[str]] = None

) -> None:

"""

Initialize the IpWareProxy class with optional proxy count and proxy list.

Args:

proxy_count: The expected number of proxies, can be None if no specific count is enforced.

proxy_list: A list of partial IP addresses as trusted proxies, can be None.

"""

if proxy_count is not None and proxy_count < 0:

raise ValueError("proxy_count must be non-negative")

self.proxy_count = proxy_count

self.proxy_list = self._validate_proxy_list(proxy_list or [])

def _validate_proxy_list(self, proxy_list: List[str]) -> List[str]:

"""

Validates that the proxy list contains only strings.

Args:

proxy_list: A list of proxies.

Returns:

The proxy list if all items are valid strings.

Raises:

ValueError: If the proxy list is not a list of strings.

"""

if not all(isinstance(ip, str) for ip in proxy_list):

raise ValueError("All elements in the proxy list must be strings.")

return proxy_list

def is_proxy_count_valid(self, ip_list: List[str], strict: bool = False) -> bool:

"""

Validates the proxy count against a list of IP addresses.

Args:

ip_list: A list of IP addresses from the request headers.

strict: If True, the number of proxies must exactly match the proxy_count.

Returns:

True if the proxy count is valid, False otherwise.

"""

if self.proxy_count is None:

return True # No proxy count specified, so always valid.

ip_count = len(ip_list)

if strict:

return ip_count - 1 == self.proxy_count

# Exact match required, excluding client's own IP.

return ip_count - 1 >= self.proxy_count

# Allow more proxies than the count, excluding client's IP.

def is_proxy_trusted_list_valid(

self, ip_list: List[str], strict: bool = False

) -> bool:

"""

Checks if all proxies in the incoming list are trusted based on the proxy_list.

Args:

ip_list: A list of IP addresses from the request headers.

strict: If True, the number of proxies must exactly match the length of proxy_list.

Returns:

True if all proxies are trusted, False otherwise.

"""

if not self.proxy_list:

return True # No specific proxies to trust, so always valid.

ip_count = len(ip_list)

proxy_list_count = len(self.proxy_list)

if strict and ip_count - 1 != proxy_list_count:

return False # Strict mode: Exact count match required.

if ip_count - 1 < proxy_list_count:

return False # Not enough IPs to match the trusted proxies.

# Verify each proxy against the trusted list by comparing each corresponding element.

return all(

str(ip).startswith(trusted_proxy_pattern)

for ip, trusted_proxy_pattern in zip(

ip_list[-proxy_list_count:], self.proxy_list

)

class IpWare(IpWareMeta, IpWareProxy, IpWareIpAddress):

"""

A class that makes a best effort to determine the client's IP address.

"""

def __init__(

self,

precedence: Optional[Tuple[str, ...]] = None,

leftmost: bool = True,

proxy_count: Optional[int] = None,

proxy_list: Optional[List[str]] = None,

) -> None:

IpWareMeta.__init__(self, precedence, leftmost)

IpWareProxy.__init__(self, proxy_count, proxy_list)

def get_meta_value(self, meta: Dict[str, str], key: str) -> str:

"""

Returns a cleaned up version of the value for a given key.

@param key: The key to look up.

@return: The value of the key or an empty string if the key is not found.

"""

meta = meta or {}

return meta.get(key, meta.get(key.replace("_", "-"), "")).strip()

def get_meta_values(self, meta: Dict[str, str]) -> List[str]:

"""

Returns a list of cleaned up values for the keys defined in 'precedence'.

@return: A list of values.

"""

meta_list: List[str] = []

for key in self.precedence:

value = self.get_meta_value(meta, key).strip()

if value:

meta_list.append(value)

return meta_list

def get_client_ip(

self,

meta: Dict[str, str],

strict: bool = False,

) -> Tuple[OptionalIpAddressType, bool]:

"""

Returns the client's IP address.

"""

loopback_list: List[IpAddressType] = []

private_list: List[OptionalIpAddressType] = []

for ip_str in self.get_meta_values(meta):

ip_list = self.get_ips_from_string(ip_str, strict)

if not ip_list:

continue

proxy_count_validated = self.is_proxy_count_valid(ip_list, strict)

if not proxy_count_validated:

continue

proxy_list_validated = self.is_proxy_trusted_list_valid(ip_list, strict)

if not proxy_list_validated:

continue

client_ip, trusted_route = self.get_best_ip(ip_list)

if client_ip is not None:

if client_ip.is_global:

return client_ip, trusted_route

if client_ip.is_loopback:

loopback_list.append(client_ip)

else:

private_list.append(client_ip)

if private_list:

return private_list[0], False

if loopback_list:

return loopback_list[0], False

return None, False

def get_best_ip(

self,

ip_list: List[IpAddressType],

) -> Tuple[OptionalIpAddressType, bool]:

"""

Determines the best possible IP address for the client from a list of IP addresses.

"""

if not ip_list:

logger.warning("Invalid IP list provided.")

return None, False

if self.proxy_list and len(self.proxy_list) > 0:

best_client_ip_index = len(self.proxy_list) + 1

best_client_ip = ip_list[-best_client_ip_index]

return best_client_ip, True

if self.proxy_count is not None:

best_client_ip_index = self.proxy_count + 1

best_client_ip = ip_list[-best_client_ip_index]

return best_client_ip, True

return ip_list[0], False

python-ipware/python_ipware/python_ipware.py at main · un33k/python-ipware · GitHub

Expand file tree

Breadcrumbs

python_ipware.py

Latest commit

History

Breadcrumbs

python_ipware.py

File metadata and controls