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
|
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# (c) 2017, Alice Ferrazzi <alice.ferrazzi@gmail.com>
# Distributed under the terms of the GNU General Public License v2 or later
import gzip
import shelve
import os
import os.path
import re
from git import Repo
from elivepatch_client.client import restful
class Kernel(object):
def __init__(self, url):
self.config = ''
self.patch = ''
self.url = url
self.kernel_version = None
self.rest_manager = restful.ManaGer(url, self.kernel_version)
self.UserID = None
def set_config(self, config_path):
self.config = config_path
def set_patch(self, patch_path):
self.patch = patch_path
def send_files(self):
# check the configuration file
path, file = (os.path.split(self.config))
f_action = FileAction(path, file)
if re.findall("[.]gz\Z", self.config):
print('gz extension')
path, file = f_action.ungz()
# if the file is .gz the configuration path is the tmp folder uncompressed config file
self.config = os.path.join(path,file)
# Get kernel version from the configuration file header
self.kernel_version = f_action.config_kernel_version(self.config)
self.rest_manager.set_kernel_version(self.kernel_version)
print('debug: kernel version =' + self.rest_manager.get_kernel_version())
path, patch_file = (os.path.split(self.patch))
# check userID
data_store = shelve.open('userid')
# get old userid if present
try:
old_userid = data_store['UserID']
except:
old_userid = None
print('no UserID')
# send uncompressed config and patch files
replay = self.rest_manager.send_file(self.config, self.patch, file, patch_file, '/elivepatch/api/v1.0/get_files')
# get userid returned from the server
userid = replay['get_config']['UserID']
self.rest_manager.set_user_id(userid)
# check if the userid is new
if userid:
try:
if userid != old_userid:
print('new userid: ' + str(userid))
data_store['UserID'] = userid
data_store.close()
except:
pass
def build_livepatch(self):
self.rest_manager.build_livepatch()
def get_livepatch(self):
self.rest_manager.get_livepatch()
class CVE(object):
def __init__(self):
self.git_url = "https://github.com/nluedtke/linux_kernel_cves"
self.repo_dir = "/tmp/kernel_cve/"
pass
def download(self):
Repo.clone_from(self.git_url, self.repo_dir)
def set_repo(self, git_url, repo_dir):
self.git_url = git_url
self.repo_dir = repo_dir
class FileAction(object):
def __init__(self, path, filename):
self.path = path
self.filename = filename
pass
def ungz(self):
path_to_store = None
get_file_path = self.path +'/'+self.filename
store_file_path = '/tmp/'+self.filename
print('file_path: '+ get_file_path)
if not os.path.isdir(get_file_path):
with gzip.open(get_file_path, 'rb') as in_file:
s = in_file.read()
# Store uncompressed file
path_to_store = store_file_path[:-3] # remove the filename extension
with open(path_to_store, 'wb') as f:
f.write(s)
print('working')
path, uncompressed_file = (os.path.split(path_to_store))
return path, uncompressed_file
def config_kernel_version(self, uncompressed_config_file):
with open(uncompressed_config_file) as f:
i = 0
while i < 2:
f.readline()
if i == 1:
kernel_line = f.readline()
i += 1
kernel_version_raw = (kernel_line.split(' ')[2])
kernel_version = kernel_version_raw.split(('-'))[0]
return kernel_version
|
GitWeb