We've been discussing how to use OpenVPN with Python in the last two articles. In this final post, I'll show how to bring it all together into a GUI with some wxPython code. I'm also going to discuss some important snippets.
The first snippet to take note of is in the run method. We need to make sure the OpenVPN service is running when we run our program or the log file won't be updated. So you'll notice that the StartService method of win32serviceutil is used. We put it in a try statement to catch the error that would occur if the OpenVPN service is already running, not found or couldn't be started. Typically, you shouldn't use a bare except as that can mask other errors in your program; however, I was unable to find the appropriate error code to use so I'll leave that to the reader.
def run(self):
"""
Run the openvpn.exe script
"""
vpnname='MCISVPN'
configfile='mcisvpn.conf'
defaultgw=''
vpnserver=''
vpnserverip = ''
print 'Starting OpenVPN Service...',
try:
win32serviceutil.StartService('OpenVPN Service', None)
except Exception, e:
print e
print 'success!'
delayedresult.startWorker(self._resultConsumer, self._resultProducer,
wargs=(self.jobID,self.abortEvent), jobID=self.jobID)
After attempting to start the OpenVPN service, I use a threading model provided by wxPython to run Golden's watcher.py code that I wrote about last time as well as keep track of my location in the log file.
The following is the main GUI code in its entirety:
from vpnTBIcon import VPNIconCtrl
import os
import sys
import Queue
import threading
import time
import win32file
import win32con
import win32serviceutil
import wx
import wx.lib.delayedresult as delayedresult
ACTIONS = {
1 : "Created",
2 : "Deleted",
3 : "Updated",
4 : "Renamed to something",
5 : "Renamed from something"
}
def watch_path (path_to_watch, include_subdirectories=False):
FILE_LIST_DIRECTORY = 0x0001
hDir = win32file.CreateFile (
path_to_watch,
FILE_LIST_DIRECTORY,
win32con.FILE_SHARE_READ | win32con.FILE_SHARE_WRITE,
None,
win32con.OPEN_EXISTING,
win32con.FILE_FLAG_BACKUP_SEMANTICS,
None
)
while 1:
results = win32file.ReadDirectoryChangesW (
hDir,
1024,
include_subdirectories,
win32con.FILE_NOTIFY_CHANGE_FILE_NAME |
win32con.FILE_NOTIFY_CHANGE_DIR_NAME |
win32con.FILE_NOTIFY_CHANGE_ATTRIBUTES |
win32con.FILE_NOTIFY_CHANGE_SIZE |
win32con.FILE_NOTIFY_CHANGE_LAST_WRITE |
win32con.FILE_NOTIFY_CHANGE_SECURITY,
None,
None
)
for action, file in results:
full_filename = os.path.join (path_to_watch, file)
if not os.path.exists (full_filename):
file_type = ""
elif os.path.isdir (full_filename):
file_type = 'folder'
else:
file_type = 'file'
yield (file_type, full_filename, ACTIONS.get (action, "Unknown"))
class Watcher (threading.Thread):
def __init__ (self, path_to_watch, results_queue, **kwds):
threading.Thread.__init__ (self, **kwds)
self.setDaemon (1)
self.path_to_watch = path_to_watch
self.results_queue = results_queue
self.start ()
def run (self):
for result in watch_path (self.path_to_watch):
self.results_queue.put (result)
# -------------------------------------------------------------------------
# GUI Code starts here
class vpnGUI(wx.App):
"""
wx application that wraps jrb's vpn script to allow it
to run in the system tray
"""
def __init__(self, redirect=False, filename=None):
wx.App.__init__(self, redirect, filename)
self.frame = wx.Frame(None, wx.ID_ANY, title='Voicemail', size=(800,500) )
self.panel = wx.Panel(self.frame, wx.ID_ANY)
self.abortEvent = delayedresult.AbortEvent()
# Set defaults
# ----------------------------------------------------------------------------------------
self.jobID = 0
# Create widget controls
# ----------------------------------------------------------------------------------------
# redirect stdout
self.log = wx.TextCtrl(self.panel, wx.ID_ANY, size=(1000,500),
style = wx.TE_MULTILINE|wx.TE_READONLY|wx.HSCROLL)
redir=RedirectText(self.log)
sys.stdout=redir
closeBtn = wx.Button(self.panel, wx.ID_ANY, 'Close')
mainSizer = wx.BoxSizer(wx.VERTICAL)
mainSizer.Add(self.log, 1, wx.ALL|wx.EXPAND, 5)
mainSizer.Add(closeBtn, 0, wx.ALL|wx.CENTER, 5)
self.panel.SetSizer(mainSizer)
# Bind events
# ----------------------------------------------------------------------------------------
self.Bind(wx.EVT_BUTTON, self.onClose, closeBtn)
self.Bind(wx.EVT_ICONIZE, self.onMinimize)
# create the system tray icon:
try:
self.tbicon = VPNIconCtrl(self.frame)
except Exception, e:
print 'Icon creation exception => %s' % e
self.tbicon = None
# comment this line out if you don't want to show the
# GUI when the program is run
self.frame.Show(True) # make the frame visible
self.run()
def run(self):
"""
Run the openvpn service
"""
vpnname='MCISVPN'
configfile='mcisvpn.conf'
defaultgw=''
vpnserver=''
vpnserverip = ''
print 'Starting OpenVPN Service...',
try:
win32serviceutil.StartService('OpenVPN Service', None)
except Exception, e:
print e
print 'success!'
delayedresult.startWorker(self._resultConsumer, self._resultProducer,
wargs=(self.jobID,self.abortEvent), jobID=self.jobID)
def _resultProducer(self, jobID, abortEvent):
"""
GUI will freeze if this method is not called in separate thread.
"""
PATH_TO_WATCH = [r'C:\Program Files\OpenVPN\log']
try: path_to_watch = sys.argv[1].split (",") or PATH_TO_WATCH
except: path_to_watch = PATH_TO_WATCH
path_to_watch = [os.path.abspath (p) for p in path_to_watch]
print "Watching %s at %s" % (", ".join (path_to_watch), time.asctime ())
files_changed = Queue.Queue ()
for p in path_to_watch:
Watcher (p, files_changed)
filepath = os.path.join(PATH_TO_WATCH[0], 'mcisvpn.log')
print 'filepath => ' + filepath
f = open(filepath)
for line in f.readlines():
print line
last_pos = f.tell()
f.close()
while not abortEvent():
try:
file_type, filename, action = files_changed.get_nowait ()
if action == 'Updated':
print 'Last pos => ', last_pos
f = open(filepath)
f.seek(last_pos)
for line in f.readlines():
if line != '\n':
print line
last_pos = f.tell()
f.close()
except Queue.Empty:
pass
time.sleep (1)
return jobID
def _resultConsumer(self, delayedResult):
jobID = delayedResult.getJobID()
assert jobID == self.jobID
try:
result = delayedResult.get()
except Exception, exc:
print "Result for job %s raised exception: %s" % (jobID, exc)
return
def onMinimize(self, event):
""" Minimize to tray """
self.frame.Hide()
def onClose(self, event):
"""
Close the program
"""
# recover stdout
sys.stdout=sys.__stdout__
# stop OpenVPN service
try:
print 'Stopping OpenVPN service...'
win32serviceutil.StopService('OpenVPN Service', None)
except Exception, e:
print e
# stop the threads
self.abortEvent.set()
# remove the icon from the tray
self.tbicon.Destroy()
# close the frame
self.frame.Close()
class RedirectText:
def __init__(self,textDisplay):
self.out=textDisplay
def write(self,string):
self.out.WriteText(string)
###### Run script! ######
if __name__ == "__main__":
app = vpnGUI()
app.MainLoop()
You'll notice that redirect stdout in the __init__ to our text control widget. To write to the widget, we use Python's print builtin. We reset stdout in the onClose event handler. This handler also stops the OpenVPN service, destroys the system tray icon and closes the program.
And that's all there really is to it. There are some links below for those of you who want to dig into these tools more.
Copyright © 2024 Mouse Vs Python | Powered by Pythonlibrary