scyther/gui/Scyther/Scyther.py

#!/usr/bin/python
"""
	Scyther : An automatic verifier for security protocols.
	Copyright (C) 2007-2020 Cas Cremers

	This program is free software; you can redistribute it and/or
	modify it under the terms of the GNU General Public License
	as published by the Free Software Foundation; either version 2
	of the License, or (at your option) any later version.

	This program 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 General Public License for more details.

	You should have received a copy of the GNU General Public License
	along with this program; if not, write to the Free Software
	Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
"""

#
# Scyther interface
#

#---------------------------------------------------------------------------

""" Import externals """
import os
import os.path
import sys
import io
import tempfile
import platform

try:
    import hashlib
    HASHLIB = True
except ImportError:
    HASHLIB = False
    pass

#---------------------------------------------------------------------------

""" Import scyther components """
from . import XMLReader
from . import Error
from . import Claim
from .Misc import *

#---------------------------------------------------------------------------

"""
Globals
"""

FirstCheck = True

#---------------------------------------------------------------------------

"""
Get current directory (for this file)
"""
def getMyDir():
    return os.path.dirname( os.path.realpath( __file__ ) )

"""
The default path for the binaries is the current one.
"""
def getBinDir():
    return getMyDir()

"""
Return Cache prefix path
Returns None if not existent
"""
def getCacheDir():
    
    tmpdir = None

    # Check if user chose the path
    cachedirkey = "SCYTHERCACHEDIR"
    if cachedirkey in list(os.environ.keys()):
        tmpdir = os.environ[cachedirkey]
        if tmpdir == "":
            # Special value: if the variable is present, but equals the empty string, we disable caching.
            return None
    else:
        # Otherwise take from path
        tmpdir = tempfile.gettempdir()
    
    # If not none, append special name
    if tmpdir != None:
        tmpdir = os.path.join(tmpdir,"Scyther-cache")
    
    # Normalize the tmpdir path.
    tmpdir = os.path.normpath(tmpdir)

    return tmpdir

    

#---------------------------------------------------------------------------

def Check():
    """
    Various dynamic checks that can be performed before starting the
    backend.
    """

    global FirstCheck

    # First time
    if FirstCheck:
        """
        Perform any checks that only need to be done the first time.
        """
        FirstCheck = False

    # Every time
    
    # Check Scyther backend program availability
    program = getScytherBackend()
    CheckSanity(program)


#---------------------------------------------------------------------------

def CheckSanity(program):
    """
    This is where the existence is checked of the Scyther backend.
    """

    if not os.path.isfile(program):
        raise Error.BinaryError(program)

#---------------------------------------------------------------------------

def EnsureString(x,sep=" "):
    """
    Takes a thing that is either a list or a string.
    Turns it into a string. If it was a list, <sep> is inserted, and the
    process iterats.

    TODO does not accept unicode yet, that is something that must be
    handled to or we run into wxPython problems eventually.
    """
    if type(x) is str:
        return x

    elif type(x) is list:
        newlist = []
        for el in x:
            newlist.append(EnsureString(el,sep))
        return sep.join(newlist)

    else:
        raise Error.StringListError(x)


#---------------------------------------------------------------------------

def getScytherBackend():
    # Where is my executable?
    #
    # Auto-detect platform and infer executable name from that
    #
    if "linux" in sys.platform:

        """ linux """
        scythername = "scyther-linux"

    elif "darwin" in sys.platform:

        """ OS X """
        # Check if there is an ARM version available at scyther-mac-arm
        # Otherwise, just fallback to the default scyther-mac which is the
        # Intel version for backwards-compatibility reasons.
        if platform.processor().startswith("arm"):
            scythername = "scyther-mac-arm"
        else:
            scythername = "scyther-mac"

    elif sys.platform.startswith('win'):

        """ Windows """
        scythername = "scyther-w32.exe"

    else:

        """ Unsupported"""
        raise Error.UnknownPlatformError(sys.platform)

    program = os.path.join(getBinDir(),scythername)
    return program


#---------------------------------------------------------------------------

class Scyther(object):
    def __init__ ( self):

        # Init
        self.program = getScytherBackend()
        self.spdl = None
        self.inputfile = None
        self.filenames = []
        self.options = ""
        self.claims = None
        self.errors = None
        self.errorcount = 0
        self.warnings = None
        self.run = False
        self.output = None
        self.cmd = None

        # defaults
        self.xml = True     # this results in a claim end, otherwise we simply get the output

    def setInput(self,spdl):
        self.spdl = spdl
        self.inputfile = None
        self.guessFileNames()

    def setFile(self,filename):
        self.inputfile = filename
        self.filenames = [self.inputfile]
        self.spdl = ""
        fp = open(filename,"r")
        for l in fp.readlines():
            self.spdl += l
        fp.close()

    def addFile(self,filename):
        self.inputfile = None
        if not self.spdl:
            self.spdl = ""
        fp = open(filename,"r")
        for l in fp.readlines():
            self.spdl += l
        fp.close()
        self.guessFileNames()

    def guessFileNames(self,spdl=None):
        """
        Try to extract filenames (well, actually, protocol names) sloppily from some spdl script.

        There are two modes:

        [init] : If the spdl parameter is empty or None, we reset the filenames and extract from self.spdl
        [add]  : If the spdl parameter is non-empty, add the extracted filenames to an existing list

        """

        if (spdl == None) or (len(spdl) == 0):
            spdl = self.spdl
            if spdl == None:
                spdl = ""
            self.filenames = []

        for sl in spdl.splitlines():
            l = sl.strip()
            prefix = "protocol "
            postfix = "("
            x = l.find(prefix)
            if x >= 0:
                # The prefix occurs
                y = l.find(postfix,x+len(prefix))
                if y >= 0:
                    gn = l[x+len(prefix):y]
                    # check for helper protocols
                    if not gn.startswith("@"):
                        if gn not in self.filenames:
                            self.filenames.append(gn)

    def addArglist(self,arglist):
        for arg in arglist:
            self.options += " %s" % (arg)

    def doScytherCommand(self, spdl, args, checkKnown=False, storePopen=None):
        """
        Cached version of the 'real' below

        TODO: CC: One possible problem with the caching is the side-effect, e.g., scyther writing to specific named output files. These are not
        captured in the cache. I don't have a good solution for that yet.
        """
        global HASHLIB

        # Can we use the cache?
        canCache = False
        if HASHLIB:
            cacheDir = getCacheDir()
            if cacheDir != None:
                canCache = True
        else:
            cacheDir = None

        # If we cannot use the cache, we either need to compute or, if checking for cache presense,...
        if not canCache:
            if checkKnown == True:
                # not using the cache, so we don't have it already
                return False
            else:
                # Need to compute
                return self.doScytherCommandReal(spdl,args, storePopen=storePopen)

        # Apparently we are supporsed to be able to use the cache
        m = hashlib.sha256()
        
        def muppet(m, s):
            m.update(s.encode('utf-8'))

        if spdl == None:
            muppet(m, "[spdl:None]")
        else:
            muppet(m, spdl)
        if args == None:
            muppet(m, "[args:None]")
        else:
            muppet(m, args)

        uid = m.hexdigest()

        # Split the uid to make 256 subdirectories with 256 subdirectories...
        prefixlen = 2
        uid1 = uid[:prefixlen]
        uid2 = uid[prefixlen:prefixlen+2]
        uid3 = uid[prefixlen+2:]

        # Possibly we could also decide to store input and arguments in the cache to analyze things later

        # Construct: cachePath/uid1/uid2/...
        path = os.path.join(cacheDir,uid1,uid2)
        name1 = "%s.out" % (uid3)
        name2 = "%s.err" % (uid3)

        fname1 = os.path.join(path, name1)
        fname2 = os.path.join(path, name2)

        try:
            """
            Try to retrieve the result from the cache
            """
            fh1 = open(fname1,"r")
            out = fh1.read()
            fh1.close()
            fh2 = open(fname2,"r")
            err = fh2.read()
            fh2.close()
            if checkKnown == True:
                # We got to here, so we have it
                return True
            else:
                # Not checking cache, we need the result
                return (out,err)
        except:
            pass

        """
        Something went wrong, do the real thing and cache afterwards
        """
        if checkKnown == True:
            # We were only checking, abort
            return False

        (out,err) = self.doScytherCommandReal(spdl,args, storePopen=storePopen)

        try:
            # Try to store result in cache
            ensurePath(path)

            fh1 = open(fname1,"w")
            fh1.write(out)
            fh1.close()

            fh2 = open(fname2,"w")
            fh2.write(err)
            fh2.close()
        except:
            pass

        return (out,err)


    def doScytherCommandReal(self, spdl, args, storePopen=None):
        """ 
        Run Scyther backend on the input
        
        Arguments:
            spdl -- string describing the spdl text
            args -- arguments for the command-line
            storePopen -- callback function to register Popen objects (used for process kill by other threads)
        Returns:
            (output,errors)
            output -- string which is the real output
            errors -- string which captures the errors
        """

        if self.program == None:
            raise Error.NoBinaryError

        # Sanitize input somewhat
        if spdl == "":
            # Scyther hickups on completely empty input
            spdl = "\n"

        # Extract filenames for error reporting later
        self.guessFileNames(spdl=spdl)

        # Generate temporary files for the output.
        (fde,fne) = tempfile.mkstemp()  # errors
        (fdo,fno) = tempfile.mkstemp()  # output
        
        # Normalize the temporary paths.
        fne = os.path.normpath(fne)
        fno = os.path.normpath(fno)
        
        if spdl:
            (fdi,fni) = tempfile.mkstemp()  # input
            fni = os.path.normpath(fni)

            # Write (input) file
            fhi = os.fdopen(fdi,'w+')
            fhi.write(spdl)
            fhi.close()

        # Generate command line for the Scyther process
        self.cmd = ""
        self.cmd += "\"%s\"" % self.program
        self.cmd += " --append-errors=\"%s\"" % fne
        self.cmd += " --append-output=\"%s\"" % fno
        self.cmd += " %s" % args
        if spdl:
            self.cmd += " \"%s\"" % fni

        # Only for debugging, really
        # print( self.cmd )

        # Start the process
        safeCommand(self.cmd, storePopen=storePopen)

        # reseek
        fhe = os.fdopen(fde)
        fho = os.fdopen(fdo)
        errors = fhe.read()
        output = fho.read()

        # clean up files
        fhe.close()
        fho.close()
        os.remove(fne)
        os.remove(fno)
        if spdl:
            os.remove(fni)

        return (output,errors)

    def sanitize(self):
        """ Sanitize some of the input """
        self.options = EnsureString(self.options)

    def verify(self,extraoptions=None,checkKnown=False,storePopen=None):
        """ Should return a list of results """
        """ If checkKnown == True, we do not call Scyther, but just check the cache, and return True iff the result is in the cache """

        # Cleanup first
        self.sanitize()
    
        # prepare arguments
        args = ""
        if self.xml:
            args += " --dot-output --xml-output --plain"
        args += " %s" % self.options
        if extraoptions:
            # extraoptions might need sanitizing
            args += " %s" % EnsureString(extraoptions)

        # Are we only checking the cache?
        if checkKnown == True:
            return self.doScytherCommand(self.spdl, args, checkKnown=checkKnown, storePopen=storePopen)

        # execute
        (output,errors) = self.doScytherCommand(self.spdl, args, storePopen=storePopen)
        self.run = True

        # process errors
        self.errors = []
        self.warnings = []
        for l in errors.splitlines():
            line = l.strip()
            if len(line) > 0:
                # filter out any non-errors (say maybe only claim etc) and count
                # them.
                if line.startswith("claim\t"):
                    # Claims are lost, reconstructed from the XML output
                    pass
                elif line.startswith("warning"):
                    # Warnings are stored seperately
                    self.warnings.append(line)
                else:
                    # otherwise it is an error
                    self.errors.append(line)

        self.errorcount = len(self.errors)
        if self.errorcount > 0:
            raise Error.ScytherError(self.errors,filenames=self.filenames,options=self.options)

        # process output
        self.output = output
        self.validxml = False
        self.claims = []
        if self.xml:
            if len(output) > 0:
                if output.startswith("<scyther>"):

                    # whoohee, xml
                    self.validxml = True

                    xmlfile = io.StringIO(output)
                    reader = XMLReader.XMLReader()
                    self.claims = reader.readXML(xmlfile)

        # Determine what should be the result
        if self.xml:
            return self.claims
        else:
            return self.output

    def verifyOne(self,cl=None,checkKnown=False,storePopen=None):
        """
        Verify just a single claim with an ID retrieved from the
        procedure below, 'scanClaims', or a full claim object

        If checkKnown is True, return if the result is already known (but never recompute).
        """
        if cl:
            # We accept either a claim or a claim id
            if isinstance(cl,Claim.Claim):
                cl = cl.id
            return self.verify("--filter=%s" % cl, checkKnown=checkKnown,storePopen=storePopen)
        else:
            # If no claim, then its just normal verification
            return self.verify(checkKnown=checkKnown,storePopen=storePopen)

    def scanClaims(self):
        """
        Retrieve the list of claims. Of each element (a claim), claim.id
        can be passed to --filter=X or 'verifyOne' later.
        A result of 'None' means that some errors occurred.
        """
        self.verify("--scan-claims")
        if self.errorcount > 0:
            return None
        else:
            self.validxml = False   # Signal that we should not interpret the output as XML
            return self.claims

    def getClaim(self,claimid):
        if self.claims:
            for cl in self.claims:
                if cl.id == claimid:
                    return cl
        return None

    def __str__(self):
        if self.run:
            if self.errorcount > 0:
                return "%i errors:\n%s" % (self.errorcount, "\n".join(self.errors))
            else:
                if self.xml and self.validxml:
                    s = "Verification results:\n"
                    for cl in self.claims:
                        s += str(cl) + "\n"
                    return s
                else:
                    return self.output
        else:
            return "Scyther has not been run yet."

#---------------------------------------------------------------------------

def GetClaims(filelist, filterlist=None):
    """
    Given a list of file names in filelist,
    returns a dictionary of filenames to lists claim names.
    Filenames which yielded no claims are filtered out.
    
    Filterlist may be None or a list of claim names (Secret, SKR, Niagree etc).
    """

    dict = {}
    for fname in filelist:
        try:
            sc = Scyther()
            sc.setFile(fname)
            l = sc.scanClaims()
            if l != None:
                cl = []
                for claim in l:
                    if filterlist == None:
                        cl.append(claim.id)
                    else:
                        if claim.claimtype in filterlist:
                            cl.append(claim.id)
                dict[fname] = cl
        except:
            pass
    return dict

#---------------------------------------------------------------------------

def FindProtocols(path="",filterProtocol=None):
    """
    Find a list of protocol names

    Note: Unix only! Will not work under windows.
    """

    import subprocess

    cmd = "find %s -iname '*.spdl'" % (path)
    plist = subprocess.getoutput(cmd).splitlines()
    nlist = []
    for prot in plist:
        if filterProtocol != None:
            if filterProtocol(prot):
                nlist.append(prot)
        else:
            nlist.append(prot)
    return nlist

#---------------------------------------------------------------------------

def GetInfo(html=False):
    """
    Retrieve a tuple (location,string) with information about the tool,
    retrieved from the --expert --version data
    """

    program = getScytherBackend()
    arg = "--expert --version"
    sc = Scyther()
    (output,errors) = sc.doScytherCommand(spdl=None, args=arg)
    if not html:
        return (program,output)
    else:
        sep = "<br>\n"
        html = "Backend: %s%s" % (program,sep)
        for l in output.splitlines():
            l.strip()
            html += "%s%s" % (l,sep)
        return html


def GetLicense():
    """
    Retrieve license information.
    """

    program = getScytherBackend()
    arg = "--license"
    sc = Scyther()
    (output,errors) = sc.doScytherCommand(spdl=None, args=arg)
    return output


#---------------------------------------------------------------------------

# vim: set ts=4 sw=4 et list lcs=tab\:>-: