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
|
# Copyright 2023 VyOS maintainers and contributors <maintainers@vyos.io>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library. If not, see <http://www.gnu.org/licenses/>.
import os
from vyos.utils.permission import chown
def makedir(path, user=None, group=None):
if os.path.exists(path):
return
os.makedirs(path, mode=0o755)
chown(path, user, group)
def file_is_persistent(path):
import re
location = r'^(/config|/opt/vyatta/etc/config)'
absolute = os.path.abspath(os.path.dirname(path))
return re.match(location,absolute)
def read_file(fname, defaultonfailure=None):
"""
read the content of a file, stripping any end characters (space, newlines)
should defaultonfailure be not None, it is returned on failure to read
"""
try:
""" Read a file to string """
with open(fname, 'r') as f:
data = f.read().strip()
return data
except Exception as e:
if defaultonfailure is not None:
return defaultonfailure
raise e
def write_file(fname, data, defaultonfailure=None, user=None, group=None, mode=None, append=False):
"""
Write content of data to given fname, should defaultonfailure be not None,
it is returned on failure to read.
If directory of file is not present, it is auto-created.
"""
dirname = os.path.dirname(fname)
if not os.path.isdir(dirname):
os.makedirs(dirname, mode=0o755, exist_ok=False)
chown(dirname, user, group)
try:
""" Write a file to string """
bytes = 0
with open(fname, 'w' if not append else 'a') as f:
bytes = f.write(data)
chown(fname, user, group)
chmod(fname, mode)
return bytes
except Exception as e:
if defaultonfailure is not None:
return defaultonfailure
raise e
def read_json(fname, defaultonfailure=None):
"""
read and json decode the content of a file
should defaultonfailure be not None, it is returned on failure to read
"""
import json
try:
with open(fname, 'r') as f:
data = json.load(f)
return data
except Exception as e:
if defaultonfailure is not None:
return defaultonfailure
raise e
def chown(path, user, group):
""" change file/directory owner """
from pwd import getpwnam
from grp import getgrnam
if user is None or group is None:
return False
# path may also be an open file descriptor
if not isinstance(path, int) and not os.path.exists(path):
return False
uid = getpwnam(user).pw_uid
gid = getgrnam(group).gr_gid
os.chown(path, uid, gid)
return True
def chmod(path, bitmask):
# path may also be an open file descriptor
if not isinstance(path, int) and not os.path.exists(path):
return
if bitmask is None:
return
os.chmod(path, bitmask)
def chmod_600(path):
""" Make file only read/writable by owner """
from stat import S_IRUSR, S_IWUSR
bitmask = S_IRUSR | S_IWUSR
chmod(path, bitmask)
def chmod_750(path):
""" Make file/directory only executable to user and group """
from stat import S_IRUSR, S_IWUSR, S_IXUSR, S_IRGRP, S_IXGRP
bitmask = S_IRUSR | S_IWUSR | S_IXUSR | S_IRGRP | S_IXGRP
chmod(path, bitmask)
def chmod_755(path):
""" Make file executable by all """
from stat import S_IRUSR, S_IWUSR, S_IXUSR, S_IRGRP, S_IXGRP, S_IROTH, S_IXOTH
bitmask = S_IRUSR | S_IWUSR | S_IXUSR | S_IRGRP | S_IXGRP | \
S_IROTH | S_IXOTH
chmod(path, bitmask)
def makedir(path, user=None, group=None):
if os.path.exists(path):
return
os.makedirs(path, mode=0o755)
chown(path, user, group)
def wait_for_inotify(file_path, pre_hook=None, event_type=None, timeout=None, sleep_interval=0.1):
""" Waits for an inotify event to occur """
if not os.path.dirname(file_path):
raise ValueError(
"File path {} does not have a directory part (required for inotify watching)".format(file_path))
if not os.path.basename(file_path):
raise ValueError(
"File path {} does not have a file part, do not know what to watch for".format(file_path))
from inotify.adapters import Inotify
from time import time
from time import sleep
time_start = time()
i = Inotify()
i.add_watch(os.path.dirname(file_path))
if pre_hook:
pre_hook()
for event in i.event_gen(yield_nones=True):
if (timeout is not None) and ((time() - time_start) > timeout):
# If the function didn't return until this point,
# the file failed to have been written to and closed within the timeout
raise OSError("Waiting for file {} to be written has failed".format(file_path))
# Most such events don't take much time, so it's better to check right away
# and sleep later.
if event is not None:
(_, type_names, path, filename) = event
if filename == os.path.basename(file_path):
if event_type in type_names:
return
sleep(sleep_interval)
def wait_for_file_write_complete(file_path, pre_hook=None, timeout=None, sleep_interval=0.1):
""" Waits for a process to close a file after opening it in write mode. """
wait_for_inotify(file_path,
event_type='IN_CLOSE_WRITE', pre_hook=pre_hook, timeout=timeout, sleep_interval=sleep_interval)
|
