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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
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
|
# Copyright (c) 2016 Hochikong
import re
from pexpect import pxssh
class VyOSError(Exception):
pass
class ConfigError(VyOSError):
pass
class CommitError(ConfigError):
pass
class ConfigLocked(CommitError):
pass
class Router(object):
def __init__(self, address, user, password):
"""Initial a router object
:param address: Router address,example:'192.168.10.10'
:param user: Router user
:param password: Router user's password
"""
self.__address = address
self.__user = user
self.__password = password
# Session flags
self.__logged_in = False
self.__session_modified = False
self.__session_saved = True
self.__conf_mode = False
self.__conn = pxssh.pxssh()
# String codec, hardcoded for now
self.__codec = "utf8"
def __execute_command(self, command):
"""This method used for sending configuration to VyOS
:param command: The configuration command
"""
self.__conn.sendline(command)
if not self.__conn.prompt():
raise VyOSError("Connection timed out")
output = self.__conn.before
# XXX: In python3 it's bytes rather than str
if isinstance(output, bytes):
output = output.decode(self.__codec)
return output
def _status(self):
"""Check the router object inner status
:return: A python dictionary include the status of the router object
"""
return {"logged_in": self.__logged_in,
"session_modified": self.__session_modified,
"session_saved": self.__session_saved,
"conf_mode": self.__conf_mode}
def login(self):
"""Login the router
"""
self.__conn.login(self.__address, self.__user, self.__password)
self.__logged_in = True
def logout(self):
"""Logout the router
"""
if not self.__logged_in:
raise VyOSError("Not logged in")
else:
if self.__conf_mode:
raise VyOSError("Cannot logout before exiting configuration mode")
else:
self.__conn.close()
self.__logged_in = False
self.__conn = pxssh.pxssh()
def configure(self):
"""Enter the VyOS configure mode
"""
if not self.__logged_in:
raise VyOSError("Cannot enter configuration mode when not logged in")
else:
if self.__conf_mode:
raise VyOSError("Session is already in configuration mode")
else:
# configure changes the prompt (from $ to #), so this is
# a bit of a special case, and we use pxssh directly instead
# of the __execute_command wrapper...
self.__conn.sendline("configure")
# XXX: set_unique_prompt() after this breaks things, for some reason
# We should find out why.
self.__conn.PROMPT = "[#$]"
if not self.__conn.prompt():
raise VyOSError("Entering configure mode failed (possibly due to timeout)")
# self.__conn.set_unique_prompt()
self.__conf_mode = True
# XXX: There should be a check for operator vs. admin
# mode and appropriate exception, but pexpect doesn't work
# with operator's overly restricted shell...
def commit(self):
"""Commit the configuration changes
"""
if not self.__conf_mode:
raise VyOSError("Cannot commit without entering configuration mode")
else:
if not self.__session_modified:
raise ConfigError("No configuration changes to commit")
else:
output = self.__execute_command("commit")
if re.search(r"Commit\s+failed", output):
raise CommitError(output)
if re.search(r"another\s+commit\s+in\s+progress", output):
raise ConfigLocked("Configuration is locked due to another commit in progress")
self.__session_modified = False
self.__session_saved = False
def save(self):
"""Save the configuration after commit
"""
if not self.__conf_mode:
raise VyOSError("Cannot save when not in configuration mode")
elif self.__session_modified:
raise VyOSError("Cannot save when there are uncommited changes")
else:
self.__execute_command("save")
self.__session_saved = True
def exit(self, force=False):
"""Exit VyOS configure mode
:param force: True or False
"""
if not self.__conf_mode:
pass
else:
# XXX: would be nice to simplify these conditionals
if self.__session_modified:
if not force:
raise VyOSError("Cannot exit a session with uncommited changes, use force flag to discard")
else:
self.__execute_command("exit discard")
self.__conf_mode = False
return
elif (not self.__session_saved) and (not force):
raise VyOSError("Cannot exit a session with unsaved changes, use force flag to ignore")
else:
self.__execute_command("exit")
self.__conf_mode = False
def set(self, path):
"""Basic 'set' method,execute the set command in VyOS
:param path: A configuration string.
e.g. 'protocols static route ... next-hop ... distance ...'
"""
if not self.__conf_mode:
raise ConfigError("Cannot execute set commands when not in configuration mode")
else:
output = self.__execute_command("{0} {1}". format("set", path))
if re.search(r"Set\s+failed", output):
raise ConfigError(output)
elif re.search(r"already exists", output):
raise ConfigError("Configuration path already exists")
self.__session_modified = True
def delete(self, path):
"""Basic 'delete' method,execute the delete command in VyOS
:param path: A configuration string.
e.g. 'protocols static route ... next-hop ... distance ...'
"""
if not self.__conf_mode:
raise ConfigError("Cannot execute delete commands when not in configuration mode")
else:
output = self.__execute_command("{0} {1}". format("delete", path))
if re.search(r"Nothing\s+to\s+delete", output):
raise ConfigError(output)
self.__session_modified = True
|
