-
Notifications
You must be signed in to change notification settings - Fork 32
Expand file tree
/
Copy pathuser_ip.py
More file actions
64 lines (53 loc) · 1.93 KB
/
user_ip.py
File metadata and controls
64 lines (53 loc) · 1.93 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
from __future__ import absolute_import
import contextlib
import logging
import string
import time
from six.moves.urllib_request import urlopen
VALID_IP_CHARACTERS = string.hexdigits + '.:'
class UserIP(object):
def __init__(self, bless_cache, maxcachetime, ip_urls, fixed_ip=False):
self.fresh = False
self.currentIP = None
self.cache = bless_cache
self.maxcachetime = maxcachetime
self.ip_urls = ip_urls
if fixed_ip:
self.currentIP = fixed_ip
self.fresh = True
def getIP(self):
if self.fresh and self.currentIP:
return self.currentIP
lastip = self.cache.get('lastip')
lastiptime = self.cache.get('lastipchecktime')
if lastiptime and lastiptime + self.maxcachetime > time.time():
return lastip
self._refreshIP()
return self.currentIP
def _refreshIP(self):
logging.debug("Getting current public IP")
ip = None
for url in self.ip_urls:
if ip:
break
else:
ip = self._fetchIP(url)
if not ip:
raise Exception('Could not refresh public IP')
self.currentIP = ip
self.fresh = True
self.cache.set('lastip', self.currentIP)
self.cache.set('lastipchecktime', time.time())
self.cache.save()
def _fetchIP(self, url):
try:
with contextlib.closing(urlopen(url, timeout=2)) as f:
if f.getcode() == 200:
content = f.read().decode().strip()[:40]
for c in content:
if c not in VALID_IP_CHARACTERS:
raise ValueError("Public IP response included invalid character '{}'.".format(c))
return content
except Exception:
logging.debug('Could not refresh public IP from {}'.format(url), exc_info=True)
return None