Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
N
nemu3
Project overview
Project overview
Details
Activity
Releases
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Issues
1
Issues
1
List
Boards
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Analytics
Analytics
CI / CD
Repository
Value Stream
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
nexedi
nemu3
Commits
7ebd9b77
Commit
7ebd9b77
authored
Jun 29, 2010
by
Martín Ferrari
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
subprocess module
parent
922ea7ae
Changes
4
Hide whitespace changes
Inline
Side-by-side
Showing
4 changed files
with
193 additions
and
34 deletions
+193
-34
src/netns/protocol.py
src/netns/protocol.py
+1
-1
src/netns/subprocess.py
src/netns/subprocess.py
+111
-6
t/test_core.py
t/test_core.py
+0
-23
t/test_subprocess.py
t/test_subprocess.py
+81
-4
No files found.
src/netns/protocol.py
View file @
7ebd9b77
...
...
@@ -295,7 +295,7 @@ class Server(object):
def
do_PROC_RUN
(
self
,
cmdname
):
try
:
chld
=
netns
.
subprocess
.
Subprocess
(
**
self
.
_proc
)
chld
=
netns
.
subprocess
.
Popen
(
**
self
.
_proc
)
except
BaseException
,
e
:
# FIXME
self
.
reply
(
500
,
"Failure starting process: %s"
%
str
(
e
))
self
.
_proc
=
None
...
...
src/netns/subprocess.py
View file @
7ebd9b77
#!/usr/bin/env python
# vim:ts=4:sw=4:et:ai:sts=4
import
os
,
signal
import
fcntl
,
grp
,
os
,
pickle
,
pwd
,
signal
,
sys
,
traceback
class
Subprocess
(
object
):
def
__init__
(
self
,
uid
,
gid
,
file
,
argv
,
cwd
=
None
,
env
=
None
,
class
Popen
(
object
):
"""Class that attempts to provide low-leven popen-like behaviour, with the
extra feature of being able to switch user before executing the command."""
def
__init__
(
self
,
user
,
file
,
argv
,
cwd
=
None
,
env
=
None
,
stdin
=
None
,
stdout
=
None
,
stderr
=
None
):
self
.
_pid
=
-
1
"""Check Python's subprocess.Popen for the intended behaviour. The
extra `user` argument, if not None, specifies a username to run the
command as, including its primary and secondary groups. If a numerical
UID is given, a reverse lookup is performed to find the user name and
then set correctly the groups. Note that `stdin`, `stdout`, and
`stderr` can only be integers representing file descriptors, and that
they are not closed by this class; piping should be handled
externally."""
userfd
=
[
stdin
,
stdout
,
stderr
]
sysfd
=
[
x
.
fileno
()
for
x
in
sys
.
stdin
,
sys
.
stdout
,
sys
.
stderr
]
# Verify there is no clash
assert
not
(
set
(
filter
(
None
,
userfd
))
&
set
(
filter
(
None
,
sysfd
)))
if
user
!=
None
:
if
str
(
user
).
isdigit
():
uid
=
int
(
user
)
try
:
user
=
pwd
.
getpwuid
(
uid
)[
0
]
except
:
raise
ValueError
(
"UID %d does not exist"
%
int
(
user
))
else
:
try
:
uid
=
pwd
.
getpwnam
(
str
(
user
))[
2
]
except
:
raise
ValueError
(
"User %s does not exist"
%
str
(
user
))
gid
=
pwd
.
getpwuid
(
uid
)[
3
]
groups
=
[
x
[
2
]
for
x
in
grp
.
getgrall
()
if
user
in
x
[
3
]]
(
r
,
w
)
=
os
.
pipe
()
pid
=
os
.
fork
()
if
pid
==
0
:
try
:
# Set up stdio piping
for
i
in
range
(
3
):
if
userfd
[
i
]
!=
None
:
os
.
dup2
(
userfd
[
i
],
sysfd
[
i
])
os
.
close
(
userfd
[
i
])
# Set up special control pipe
os
.
close
(
r
)
fcntl
.
fcntl
(
w
,
fcntl
.
F_SETFD
,
fcntl
.
FD_CLOEXEC
)
if
user
!=
None
:
# Change user
os
.
setgid
(
gid
)
os
.
setgroups
(
groups
)
os
.
setuid
(
uid
)
if
cwd
!=
None
:
os
.
chdir
(
cwd
)
if
not
argv
:
argv
=
[
file
]
if
'/'
in
file
:
# Should not search in PATH
if
env
!=
None
:
os
.
execve
(
file
,
argv
,
env
)
else
:
os
.
execv
(
file
,
argv
)
else
:
# use PATH
if
env
!=
None
:
os
.
execvpe
(
file
,
argv
,
env
)
else
:
os
.
execvp
(
file
,
argv
)
raise
RuntimeError
(
"Unreachable reached!"
)
except
:
try
:
(
t
,
v
,
tb
)
=
sys
.
exc_info
()
# Got the child_traceback attribute trick from Python's
# subprocess.py
v
.
child_traceback
=
""
.
join
(
traceback
.
format_exception
(
t
,
v
,
tb
))
os
.
write
(
w
,
pickle
.
dumps
(
v
))
os
.
close
(
w
)
except
:
traceback
.
print_exc
()
os
.
_exit
(
1
)
os
.
close
(
w
)
# read EOF for success, or a string as error info
s
=
""
while
True
:
s1
=
os
.
read
(
r
,
4096
)
if
s1
==
""
:
break
s
+=
s1
os
.
close
(
r
)
if
s
==
""
:
self
.
_pid
=
pid
return
# It was an error
os
.
waitpid
(
pid
,
0
)
raise
pickle
.
loads
(
s
)
@
property
def
pid
(
self
):
return
self
.
_pid
def
poll
(
self
):
"""Check if the process already died. Returns the exit code or None if
the process is still alive."""
r
=
os
.
waitpid
(
self
.
_pid
,
os
.
WNOHANG
)
if
r
[
0
]:
del
self
.
_pid
return
r
[
1
]
return
None
def
wait
(
self
):
return
0
"""Wait for process to die and return the exit code."""
r
=
os
.
waitpid
(
self
.
_pid
,
0
)[
1
]
del
self
.
_pid
return
r
def
kill
(
self
,
sig
=
signal
.
SIGTERM
):
return
self
.
_pid
"""Kill the process with the specified signal. Note that the process
still needs to be waited for to avoid zombies."""
os
.
kill
(
self
.
_pid
,
sig
)
t/test_core.py
View file @
7ebd9b77
...
...
@@ -17,28 +17,5 @@ class TestConfigure(unittest.TestCase):
self
.
assertRaises
(
AttributeError
,
setattr
,
netns
.
config
,
'run_as'
,
-
1
)
def
test_config_run_as_runtime
(
self
):
user
=
netns
.
config
.
run_as
=
'nobody'
uid
=
pwd
.
getpwnam
(
user
)[
2
]
gid
=
pwd
.
getpwnam
(
user
)[
3
]
groups
=
[
x
[
2
]
for
x
in
grp
.
getgrall
()
if
user
in
x
[
3
]]
node
=
netns
.
Node
()
app
=
node
.
start_process
([
"sleep"
,
"1000"
])
pid
=
app
.
pid
# FIXME: non-portable *at all*
stat
=
open
(
"/proc/%d/status"
%
pid
)
while
True
:
data
=
stat
.
readline
()
fields
=
data
.
split
()
if
fields
[
0
]
==
'Uid:'
:
self
.
assertEquals
(
fields
[
1
:
4
],
(
uid
,)
*
4
)
if
fields
[
0
]
==
'Gid:'
:
self
.
assertEquals
(
fields
[
1
:
4
],
(
gid
,)
*
4
)
if
fields
[
0
]
==
'Groups:'
:
self
.
assertEquals
(
set
(
fields
[
1
:]),
set
(
groups
))
break
stat
.
close
()
if
__name__
==
'__main__'
:
unittest
.
main
()
t/test_subprocess.py
View file @
7ebd9b77
#!/usr/bin/env python
# vim:ts=4:sw=4:et:ai:sts=4
import
netns
import
unittest
import
netns
,
netns
.
subprocess
,
test_util
import
grp
,
os
,
pwd
,
signal
,
sys
,
unittest
def
_stat
(
path
):
try
:
return
os
.
stat
(
user
)
except
:
return
None
def
_getpwnam
(
user
):
try
:
return
pwd
.
getpwnam
(
user
)
except
:
return
None
def
_getpwuid
(
uid
):
try
:
return
pwd
.
getpwuid
(
uid
)
except
:
return
None
class
TestSubprocess
(
unittest
.
TestCase
):
def
test_start_process
(
self
):
pass
def
_check_ownership
(
self
,
user
,
pid
):
uid
=
pwd
.
getpwnam
(
user
)[
2
]
gid
=
pwd
.
getpwnam
(
user
)[
3
]
groups
=
[
x
[
2
]
for
x
in
grp
.
getgrall
()
if
user
in
x
[
3
]]
stat
=
open
(
"/proc/%d/status"
%
pid
)
while
True
:
data
=
stat
.
readline
()
fields
=
data
.
split
()
if
fields
[
0
]
==
'Uid:'
:
self
.
assertEquals
(
fields
[
1
:
4
],
(
uid
,)
*
4
)
if
fields
[
0
]
==
'Gid:'
:
self
.
assertEquals
(
fields
[
1
:
4
],
(
gid
,)
*
4
)
if
fields
[
0
]
==
'Groups:'
:
self
.
assertEquals
(
set
(
fields
[
1
:]),
set
(
groups
))
break
stat
.
close
()
def
setUp
(
self
):
self
.
nouid
=
65535
while
_getpwuid
(
self
.
nouid
):
self
.
nouid
-=
1
self
.
nouser
=
'foobar'
while
_getpwnam
(
self
.
nouser
):
self
.
nouser
+=
'_'
self
.
nofile
=
'/foobar'
while
_stat
(
self
.
nofile
):
self
.
nofile
+=
'_'
# XXX: unittest still cannot skip tests
#@unittest.skipUnless(os.getuid() == 0, "Test requires root privileges")
@
test_util
.
skipUnless
(
os
.
getuid
()
==
0
,
"Test requires root privileges"
)
def
test_popen_chuser
(
self
):
user
=
'nobody'
p
=
netns
.
subprocess
.
Popen
(
user
,
'/bin/sleep'
,
[
'/bin/sleep'
,
'1000'
])
self
.
_check_ownership
(
user
,
p
.
pid
)
p
.
kill
()
self
.
assertEquals
(
p
.
wait
(),
signal
.
SIGTERM
)
def
test_popen_basic
(
self
):
# User does not exist
self
.
assertRaises
(
ValueError
,
netns
.
subprocess
.
Popen
,
self
.
nouser
,
'/bin/sleep'
,
[
'/bin/sleep'
,
'1000'
])
self
.
assertRaises
(
ValueError
,
netns
.
subprocess
.
Popen
,
self
.
nouid
,
'/bin/sleep'
,
[
'/bin/sleep'
,
'1000'
])
# Invalid CWD: it is a file
self
.
assertRaises
(
OSError
,
netns
.
subprocess
.
Popen
,
None
,
'/bin/sleep'
,
None
,
cwd
=
'/bin/sleep'
)
# Invalid CWD: does not exist
self
.
assertRaises
(
OSError
,
netns
.
subprocess
.
Popen
,
None
,
'/bin/sleep'
,
None
,
cwd
=
self
.
nofile
)
# Exec failure
self
.
assertRaises
(
OSError
,
netns
.
subprocess
.
Popen
,
None
,
self
.
nofile
,
None
)
# Test that the environment is cleared: sleep should not be found
# XXX: This should be a python bug: if I don't set PATH explicitly, it
# uses a default search path
self
.
assertRaises
(
OSError
,
netns
.
subprocess
.
Popen
,
None
,
'sleep'
,
None
,
env
=
{
'PATH'
:
''
})
#p = netns.subprocess.Popen(None, '/bin/sleep', ['/bin/sleep', '1000'],
# cwd = '/', env = [])
# FIXME: tests fds
if
__name__
==
'__main__'
:
unittest
.
main
()
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment